Forem: Ahsan Nabi Dar

Building a Scalable Audit Logging Pipeline in Elixir: Handling Millions of Events Without Breaking Your Database

Ahsan Nabi Dar — Mon, 29 Dec 2025 11:43:55 +0000

In enterprise applications dealing with payroll, compensation, and benefits data, every change matters. When a single data point can cascade through multiple systems, affecting someone's salary or benefits package, having a complete audit trail isn't just nice to have—it's mission-critical. But capturing these events is only half the battle. The real challenge? Managing millions of audit records without turning your database into a bottleneck.
Let me walk you through how we built a scalable audit logging pipeline for our Elixir-based application, handling over 5 million events monthly per client while keeping our core services fast and our data accessible.

The Challenge

Our core application is built entirely in Elixir, operating in both public and corporate domains. The complexity of our data flows is significant: each data point typically touches multiple workers and goes through an average of 6-7 transactions before settling. While Elixir's concurrency model handles millions of operations with ease, audit logging presents unique challenges that can't simply be solved by throwing more processes at the problem.
The issue compounds quickly. With over 5 million audit events per month for a single client, we're looking at 60+ million records annually. Left unchecked, this growth pattern threatens to turn your primary database into a bottleneck, slowing down queries, bloating backups, and eventually impacting the performance of your core business operations.

Our Solution Architecture

We designed a pipeline that separates audit capture from audit storage, leveraging the outbox pattern to ensure reliable event delivery while keeping our operational database lean.

Architecture Flow

The Components

1. Carbonite: Capturing Changes at the Database Level
Carbonite is an Elixir library that uses database triggers to automatically capture all changes to your tables. This approach offers several advantages:

Zero application code changes: You don't need to manually log changes throughout your codebase
Complete coverage: Every INSERT, UPDATE, and DELETE is captured automatically
Transactional consistency: Audit records are created in the same transaction as the data changes
Rich metadata: Captures before/after values, timestamps, and transaction context

The beauty of Carbonite is that it operates at the database level. Even if you have complex business logic spread across multiple modules and processes, you get comprehensive audit coverage without littering your code with logging statements.

2. The Outbox Pattern: Reliable Event Publishing
The outbox pattern solves a critical problem: how do you reliably publish events from a transactional system? By writing audit events to an outbox table within the same database transaction as your business data, you ensure that either both succeed or both fail—no orphaned audit records or missed events.
Carbonite's outbox utility reads from these outbox tables and publishes events to external systems. This decouples the write path (fast database operations) from the publish path (potentially slower network operations).

3. LavinMQ: Lightweight Message Broker
We chose LavinMQ as our message broker for several reasons:

AMQP protocol support: Industry-standard messaging protocol
Lightweight and fast: Designed for high throughput with minimal resource overhead
Reliable delivery: Ensures messages aren't lost between components
Buffering capability: Handles traffic spikes gracefully

LavinMQ acts as the shock absorber in our pipeline, allowing the audit capture rate to differ from the processing rate without losing data.

4. Vector.dev: Event Processing and Routing
Vector is a high-performance observability data pipeline that consumes messages from LavinMQ and handles:

Transformation: Reshaping events into the desired format
Enrichment: Adding metadata or contextual information
Routing: Directing events to appropriate destinations
Batching: Optimizing write patterns to storage

Vector's configuration-as-code approach makes it easy to modify the pipeline without redeploying applications.

5. Backblaze: Cost-Effective Long-Term Storage
For long-term storage, we use Backblaze B2 via the S3-compatible API. Backblaze offers:

Low cost: Significantly cheaper than traditional cloud object storage
S3 compatibility: Works with existing S3 tooling and libraries
Durability: Enterprise-grade data protection
Scalability: Handles petabytes of data without configuration changes

Key Benefits of This Architecture

Decoupling Operations
By separating audit capture, transport, and storage into distinct layers, each component can be scaled, maintained, and upgraded independently. Your core application doesn't need to know or care about where audit logs ultimately end up.

Multiple Destinations
Once audit events are flowing through the pipeline, routing them to multiple destinations becomes trivial. You might:

Send recent events to a hot analytics database for real-time dashboards
Archive all events to object storage for compliance
Stream specific events to client-accessible APIs
Feed events into a data lake for data warehouse construction
Build data lineage graphs showing how values changed over time

Specialized Tooling
Each component in our pipeline is purpose-built for its role. We're not trying to make our Elixir application handle message queuing, or forcing our database to store years of historical data. We use mature protocols (AMQP, S3) and battle-tested services, reducing the surface area for bugs and operational issues.

Performance at Scale
Our operational database stays fast because we're continuously moving audit data out. LavinMQ handles traffic spikes without blocking database commits. Vector batches writes to optimize storage operations. The result is a system that handles millions of events per month without breaking a sweat.

Implementation Considerations

Monitoring and Observability
With a multi-component pipeline, observability is crucial. We monitor:

Outbox table size and processing lag
LavinMQ queue depths and consumption rates
Vector processing rates and error counts
Backblaze write success rates and latency

Data Retention and Compliance
The pipeline makes it easy to implement sophisticated retention policies. Recent data might stay in a fast query layer for 90 days, then move to cheaper storage for 7 years to meet compliance requirements, then be deleted or archived to glacier storage.

Backpressure Handling
Each component needs to handle backpressure gracefully. If Vector can't keep up with LavinMQ, messages queue in the broker. If Backblaze is slow, Vector batches and retries. The outbox pattern ensures no data is lost even if downstream systems are temporarily unavailable.

Conclusion

Building a scalable audit logging pipeline requires thinking beyond simple database inserts. By leveraging the outbox pattern with Carbonite, and building a decoupled event pipeline with LavinMQ, Vector.dev, and Backblaze, we've created a system that:

Captures every change automatically at the database level
Handles millions of events per month without impacting application performance
Provides flexibility to route audit data to multiple destinations
Uses mature, purpose-built tools for each layer
Scales horizontally without architectural changes

For applications in regulated industries or where data lineage is critical, this architecture provides the foundation for comprehensive audit logging that doesn't compromise on performance or scalability. The key is recognizing that audit logging is a data pipeline problem, not just a database problem, and designing accordingly.

Building Resilient Event-Driven Systems: Lessons from the Distributed Trenches

Ahsan Nabi Dar — Tue, 04 Nov 2025 07:49:55 +0000

When you first look at a distributed architecture diagram with services scattered across multiple cloud providers, regions, and continents, it's easy to feel overwhelmed. Network partitions, timeouts, SSL handshake failures, connection drops—the list of things that can go wrong seems endless. And they do go wrong, constantly. In our globally distributed application running across 5 regions on Fly.io, we see these failures every single day. Phoenix.PubSub disconnecting from Redis. PostgreSQL protocols timing out mid-query. RabbitMQ brokers closing connections unexpectedly. It looks scary, and honestly, it should be. But here's the thing: with the right architectural choices, the right technology stack, and a healthy respect for the fallacies of distributed computing, you can build systems that not only survive these failures but thrive because of how they handle them.

The Network Is Not Reliable (And Never Will Be)

Peter Deutsch's famous "8 Fallacies of Distributed Computing" starts with the most dangerous assumption: the network is reliable. It's not. When you're orchestrating services across Redis on Upstash, LavinMQ on CloudAMPQ, Postgres on Neon, and storage on Backblaze B2—all while serving users globally through BunnyCDN—you're essentially building on top of organized chaos. The cost of communication is never zero, and latency isn't just a number on a dashboard; it's a real constraint that shapes your entire architecture.
Looking at our Sentry error logs tells the story plainly: DBConnection.ConnectionError: ssl recv (idle): timeout, Phoenix.PubSub disconnected from Redis with reason :ssl_closed, Cannot connect to RabbitMQ broker: :timeout. These aren't exceptional circumstances; they're Tuesday. The errors show up, they resolve themselves, and the application keeps running. That's not luck—that's design.

Let It Crash: More Than Just a Slogan

Elixir's "let it crash" philosophy, inherited from Erlang's decades of building fault-tolerant telecom systems, is often misunderstood. It doesn't mean writing careless code or ignoring errors. It means designing systems where failures are isolated, supervised, and automatically recovered from. When a PostgreSQL connection times out after 4 days of idle time, the connection process crashes, and the supervisor immediately spawns a new one. When Phoenix.PubSub loses its Redis connection, it gracefully reconnects without taking down the entire application.
This is visible in our error patterns. Notice how the Postgrex protocol disconnections show up with (No error message) but are all marked as Resolved. The system detected the failure, handled it, and moved on. No manual intervention. No emergency pages at 3 AM. The BEAM VM's process isolation means that a failing database connection doesn't cascade into a system-wide outage—it's contained, logged, and recovered.

Event-Driven Architecture: Decoupling for Resilience

At the heart of our architecture lies event-driven design, and it's what allows us to scale without losing data. When a user uploads a file, that action triggers events that flow through the system asynchronously. The web request completes quickly, returning control to the user, while the actual processing happens in the background through LavinMQ for inter-service communication and Oban for background job processing.
Here's why this matters: when LavinMQ has a momentary connection issue (as we see in the logs with RuntimeError: unexpected error when connecting to RabbitMQ broker), the messages aren't lost. They're persisted in PostgreSQL through Oban's reliable job queue. The system retries with exponential backoff—starting with short delays and gradually increasing them to avoid overwhelming a recovering service. This retry strategy is visible in our error resolution times: some errors resolve in minutes, others take hours, but they all resolve without manual intervention.
Using Postgres as Oban's backend is a deliberate choice. While LavinMQ handles the real-time, high-throughput message passing between services, Oban manages critical background tasks that absolutely cannot be lost. Database-backed job queues give us transactional guarantees—if the job is enqueued, it will eventually be processed, even if it takes multiple retries across connection failures.

The Technology Stack: Intentional Choices

Every piece of our stack was chosen with resilience and global distribution in mind:
Fly.io provides globally distributed compute with applications running in 5 regions simultaneously. When a region has issues, traffic automatically routes to healthy regions. The edge-first architecture means users always connect to the nearest available instance.
Upstash Redis gives us globally replicated cache and pub/sub capabilities. Even when connections drop (as they inevitably do), the application degrades gracefully, fetching from the database instead of serving stale cache or crashing.
CloudAMPQ's LavinMQ offers a lightweight, high-performance message broker. Its RabbitMQ compatibility means we get the proven AMQP protocol with better performance characteristics. Message persistence ensures that even during connection issues, messages wait in queues rather than disappearing into the void.
Neon's Postgres provides serverless Postgres with branching and point-in-time recovery. The connection pooling and automatic scaling mean we can handle traffic spikes without manual database provisioning. When we see those ssl recv (idle): timeout errors, it's often because a connection has been idle during low-traffic periods—Neon's serverless nature shuts down idle resources, and our connection pooler handles reconnection transparently.
Backblaze B2 with S3 protocol gives us cost-effective, reliable object storage. The S3 compatibility means we can use battle-tested client libraries, and the global CDN integration through BunnyCDN ensures low-latency access worldwide.
CircleCI handles our continuous deployment pipeline, automatically deploying code changes across all regions. This is crucial because fixing issues often means deploying new code, and we need that process to be fast and reliable.
Sentry is our observability layer, showing us not just when things fail, but how they fail and how they recover. The error patterns in Sentry guided many of our retry strategies and timeout configurations.

The Reactive Manifesto in Practice

Our architecture embodies the principles of the Reactive Manifesto—responsive, resilient, elastic, and message-driven:
Responsive: Users get fast responses because we don't block on slow operations. File uploads return immediately while processing happens asynchronously.
Resilient: Failures are isolated and don't cascade. A database timeout doesn't crash the application; it crashes a single connection process that's immediately restarted.
Elastic: Our globally distributed compute and serverless database scale up and down based on demand without manual intervention.
Message-driven: Events flow through message queues, decoupling services and allowing them to process work at their own pace while maintaining backpressure.

Practical Patterns for Resilience

Several concrete patterns emerge from our experience:
Exponential backoff with jitter spaces out retries intelligently. The first retry happens quickly (maybe the network hiccup was momentary), but subsequent retries back off exponentially, with random jitter to prevent thundering herd problems.
Dead letter queues capture messages that fail repeatedly after all retries. This prevents poison messages from blocking queue processing while ensuring we can investigate and manually recover them later.
Idempotent operations ensure that retries don't cause duplicate side effects. When LavinMQ redelivers a message after a connection issue, we can safely process it again without corrupting data.
Graceful degradation means falling back to reduced functionality rather than failing completely. If Redis is down, we serve from the stale in memory cache rather than throwing errors.

The Cost of Resilience

Building resilient distributed systems isn't free. There's operational complexity—more services to monitor, more failure modes to understand, more logs to analyze. There's performance overhead—retries mean higher latency for failed operations, circuit breakers mean some requests get rejected that might have eventually succeeded.
But the alternative is worse. Without proper retry logic, timeout handling, and failure isolation, that DBConnection.ConnectionError doesn't just log to Sentry and resolve itself—it crashes your application. That momentary Redis connection drop doesn't trigger a graceful reconnection—it takes down your real-time features until someone manually restarts the service.
Our error logs show the price we pay for resilience: hundreds of timeout errors, connection failures, and protocol disconnections every week. But they also show something more important: they all resolve themselves. The system heals, automatically, without human intervention. That's what resilience looks like in production.

Conclusion: Embrace the Chaos

Distributed systems are inherently chaotic. Networks partition, services timeout, connections close unexpectedly. You can't prevent these failures, but you can design for them. Elixir's supervisor trees give you automatic recovery. Event-driven architecture gives you decoupling and scalability. Proper retry strategies give you resilience. Message queues give you durability.
When we look at our Sentry dashboard and see all those resolved errors—PostgreSQL timeouts, Redis disconnections, RabbitMQ failures—we don't see problems. We see evidence that the system is working as designed. Each of those errors represents a moment where the system detected a failure, handled it gracefully, and recovered automatically.
Building distributed systems is still scary. The architecture diagrams are still complex. The failure modes are still numerous. But with the right tools, the right patterns, and a healthy respect for the network's unreliability, you can build systems that are truly resilient—systems that bend but don't break, that stumble but don't fall, that meet the challenges of global distribution head-on and emerge stronger for it.
The network will fail. Your services will timeout. Connections will drop. Build accordingly.

Bridging the AI Gap: Simplifying LLM Development in Elixir with langchain

Ahsan Nabi Dar — Thu, 01 May 2025 16:11:57 +0000

Artificial Intelligence is evolving at a breakneck pace. Every week, it feels like there’s a new breakthrough, a novel architecture, or yet another company launching its own large language model (LLM). With so much innovation happening simultaneously, keeping up with the ever-changing landscape can feel overwhelming — even for seasoned developers.

One of the biggest challenges developers face today is inconsistency across platforms. While OpenAI pioneered the API format that many now follow, major inference service providers such as Anthropic, Hugging Face, and Google Cloud have each developed their own unique interfaces. This fragmentation makes it difficult to switch between models or providers without rewriting significant chunks of code.

This is where LangChain comes into play. Originally built for Python, LangChain quickly became the go-to framework for working with LLMs. It introduced a standardized way to interact with different models and APIs, abstracting away the complexity and letting developers focus on building powerful applications.

But what if you're an Elixir developer? That’s where the Elixir community steps in with langchain — an Elixir port of the LangChain framework.

Created by members of the Elixir ecosystem, langchain brings the same power and flexibility of LangChain to Elixir developers. It provides a consistent abstraction layer over various inference providers, allowing you to seamlessly integrate with OpenAI, Anthropic, Hugging Face, and more — all without worrying about the nuances of each provider’s API.

Recently, I was working with Google Gemini models via their Vertex AI API . Initially, I had hand-rolled the HTTP requests using Google’s API format. But after discovering langchain, I decided to refactor my implementation to use the framework instead.

However, I hit a roadblock — file URL support wasn’t implemented yet for Vertex AI .

Now here’s where things get interesting:

While Google AI Studio only allows file URLs uploaded through its own file service, Vertex AI supports processing media from any publicly crawlable third-party URL — a very useful feature for real-world applications where files might live outside of Google’s ecosystem.

At the time, the Elixir community had just added file URL support as recent as last month for Google AI Studio , likely because it’s the more commonly used platform among casual users. But Vertex AI , popular among enterprise developers and production setups, didn't yet have this capability in langchain.

So I decided to step in.

Building on top of the great work already done by the community, I submitted a PR to add full file URL support for Gemini models via Vertex AI in langchain — and good news: it got merged!

This enhancement now enables developers to:

Pass public URLs directly to Gemini models via Vertex AI
Process images, PDFs, and other media formats hosted externally
Leverage the full capabilities of Gemini within the Elixir ecosystem

Without langchain the implementation to Vertex AI looks as such where you need to maintain the request format and which will change for each provider.

 defp inference(prompt, mime_type, content_url) do
     json = %{
       "contents" => %{
         "role" => "user",
         "parts" => [
           %{
             "fileData" => %{
               "mimeType" => mime_type,
               "fileUri" => content_url
             }
           },
           %{
             "text" => prompt
           }
         ]
       },
       "generationConfig" => %{
         "temperature" => 1.0,
         "topP" => 0.8,
         "topK" => 10
       }
     }

     Req.post!(vertex_endpoint(),
       auth: {:bearer, auth_token()},
       json: json,
       receive_timeout: @timeout
     ).body
     |> case do
       %{"candidates" => [%{"content" => %{"parts" => [%{"text" => response}]}}]} -> {:ok, response}

       %{"error" => %{
           "code" => code,
            "details" => _details,
           "message" => message,
           "status" => status
         }
       } ->
         msg = "GCloud Vertex API error {status} ({code}), {message}"
         Logger.info(msg)
         {:error, msg}
     end
   end

with Elixir langchain


 defp inference(prompt, mime_type, content_url) do
    config = %{
      model: model_id(),
      endpoint: vertex_endpoint(),
      api_key: auth_token(),
      temperature: 1.0,
      top_p: 0.8,
      top_k: 10,
      receive_timeout: @timeout
    }

    content = %{
      mime_type: mime_type,
      url: content_url
    }

    vertex_ai(config, prompt, content)
    |> case do
      {:ok, response} ->
        {:ok, response}

      {:error, changeset} ->
        {:error, changeset}
    end
  end

def vertex_ai(config, prompt, content) do
    model = ChatVertexAI.new!(config)

    %{llm: model, verbose: false, stream: false}
    |> LLMChain.new!()
    |> LLMChain.add_message(
      Message.new_user!([
        ContentPart.new!(%{type: :text, content: prompt}),
        ContentPart.new!(%{
          type: :file_url,
          content: content.url,
          options: [media: content.mime_type]
        })
      ])
    )
    |> LLMChain.run()
    |> parse_content()
  end

  defp parse_content(
         {:ok,
          %LangChain.Chains.LLMChain{
            llm: _llm,
            last_message: %LangChain.Message{
              content: [
                %LangChain.Message.ContentPart{type: :text, content: content}
              ]
            }
          }}
       ) do

    {:ok, content}
  end

  defp parse_content({:error, changeset}) do
    Logger.info(changeset |> inspect())
    {:error, "Langchain encountered error !!!!"}
  end

As AI continues to evolve, tools like LangChain and its Elixir counterpart, langchain, are essential for developers looking to harness the power of LLMs without getting bogged down by platform-specific quirks.

Whether you're building chatbots, data analyzers, or intelligent agents, langchain empowers you to stay agile, innovative, and ahead of the curve — all while staying true to the elegance and performance of Elixir.

So if you’re an Elixir developer diving into the world of AI, don’t reinvent the wheel. Let langchain do the heavy lifting, while you focus on creating something amazing with the power of Elixir.

Overcoming Challenges with Google Vertex API, s6-overlay, and Fly.io: A Journey to Seamless Container Orchestration

Ahsan Nabi Dar — Sun, 02 Feb 2025 16:36:55 +0000

When working with Google Vertex AI's API, one of the challenges developers face is its reliance on access tokens that are valid for only 3600 seconds (or 1 hour). Unlike other APIs that support long-lived API tokens, Vertex AI requires you to generate and refresh these short-lived access tokens. This introduces additional complexity when deploying applications, especially in containerized environments where stateless operations are preferred.

In this blog post, we'll explore how to integrate Google Vertex AI's authentication requirements into an Elixir application running inside a Docker container on Fly.io. We'll also discuss how to use S6-overlay for process orchestration, ensuring smooth execution of database migrations, GCloud CLI setup, and maintaining statelessness by avoiding hardcoding sensitive credentials like service account key files.

To authenticate with Google Vertex AI, you need to generate an access token using the gcloud CLI. This involves setting up a service account, obtaining a JSON key file, and authenticating the CLI. In a containerized environment, this setup must be automated and integrated seamlessly into your application's lifecycle.

Additionally, when deploying to platforms like Fly.io, there are specific constraints:

Fly.io Deployment Flow : After successful deployment, Fly.io runs migration commands as part of the release process.
Firecracker VMs : Fly.io containers run as lightweight virtual machines (Firecracker VMs), which can interfere with traditional container entrypoint setups like ENTRYPOINT.
These challenges necessitate a robust solution for managing processes within the container, ensuring that:

Database migrations run before the app starts.
The gcloud CLI is properly configured during the app's lifecycle.
The container remains stateless, avoiding hardcoded credentials.

For container process orchestration, I’ve come to rely on s6-overlay. It’s a lightweight, powerful tool that allows you to manage multiple processes in a container, define dependencies between them, and ensure they start and stop in the correct order. It’s perfect for scenarios where you need to run one-off tasks (like database migrations) alongside long-running processes (like an Elixir app).

In this case, I needed to set up a dependency graph like this:

migration → Elixir app → gcloud

Migration: A one-shot task that runs database migrations.

Elixir app: The main application that runs continuously.

gcloud: A one-shot task that sets up the gcloud CLI and generates the access token.

Both the migration and gcloud tasks are one-shot processes—they only need to run once per container lifecycle. The Elixir app, however, is a long-running process. If gcloud fails, it only affects the function that relies on the Google Vertex API, allowing the rest of the app to continue running.

Deploying to Fly.io added another layer of complexity. Fly.io uses Firecracker VMs to run containers, and this introduced an unexpected issue: s6-overlay wouldn’t start as usual. After some digging, I found a GitHub gist where someone had faced a similar problem and shared a solution. With a few tweaks, I was able to adapt their approach to get s6-overlay working in the Fly.io environment.

ENTRYPOINT [ \
    "unshare", "--pid", "--fork", "--kill-child=SIGTERM", "--mount-proc", \
    "perl", "-e", "$SIG{INT}=''; $SIG{TERM}=''; exec @ARGV;", "--", \
    "/init" ]

One of the key benefits of using s6-overlay was the ability to keep the container stateless. Instead of copying the gcloud keyfile into the container, I passed it as an environment variable. The s6-overlay setup script then used this variable to authenticate the gcloud CLI. This approach not only simplified the container setup but also improved security by avoiding hardcoding sensitive credentials into the container image.

Here’s a high-level overview of the final setup:

Service Account Keyfile: Stored as an environment variable in Fly.io.
s6-overlay: Used to orchestrate the processes and manage dependencies.
Migration Script: Run as a one-shot task before the Elixir app starts.
gcloud Setup: Run as a one-shot task to authenticate and generate the access token.
Elixir App: The main application, which starts after the migration and gcloud tasks complete.

The folder structure to setup the process is as such

Integrating Google Vertex API into an Elixir app deployed on Fly.io was a challenging but rewarding experience. By leveraging s6-overlay for process orchestration and adapting to Fly.io’s unique environment, I was able to create a robust, stateless deployment pipeline. If you’re facing similar challenges, I hope this post provides some inspiration and guidance.

Leveraging GenServer and Queueing Techniques: Handling API Rate Limits to AI Inference services

Ahsan Nabi Dar — Sat, 02 Nov 2024 08:38:17 +0000

In the realm of efficient application development, managing external service rate limits is a pivotal challenge. Recently faced this task while interfacing with the Fireworks serverless API. In the world of serverless APIs, rate limits can be a significant challenge to overcome. The Fireworks AI platform, in particular, comes with a shared 600 requests per minute limit between inference and embedding functionalities. However, with the right approach, it's possible to optimize this limit to accommodate multiple users and ensure consistent response times.Fireworks provides a set of 2 API keys which means you can go upto 1200 req/min if you can successfully load balance between them. I wrote a service called ping pong to do that but we won't be discussing about Load balancing. We will be going over more exciting bit of ping pong about how to manage rate limit and queue incoming requests to not drop any request using GenServer and queue them with an acceptable timeout limit.

In a typical scenario, users may submit multiple requests simultaneously, with each subsequent request consuming the available quota more quickly than the previous one. For instance, if 600 requests come in within the first ten seconds of processing, the remaining 50 seconds will see all requests being rate-limited due to their higher priority.

However, we needed a solution that could efficiently process all incoming requests while maintaining fairness and preventing any potential overloading. Our approach involved utilizing GenServer and queues to manage these requests effectively.

GenServer is a powerful tool in Elixir for managing stateful applications. It allows us to hold the request count state and queue up new requests, ensuring they are processed in an orderly manner without overwhelming the system.The built-in timeout mechanism within genservers ensures that if a wait becomes too protracted, older requests are dropped gracefully without spiraling out to form an uncontrollable backlog prone to overwhelm our system resources; maintaining the sanctity of fairness and balance.

We decided to have a separate rate limiter GenServer process for both inference and embedding functionalities due to their different priorities. For example, if there's an inferred request coming in while an embedding request is being processed, it's essential to prioritize the more frequent requests first.

In our setup, we defined a limit of 400 requests per minute for inference and 200 for embedding, knowing that inference would typically have higher frequency. This method ensures that critical or high-priority tasks receive timely attention while less demanding operations can be processed with lower priority but still within acceptable limits.

When there's no requests in the queue, it’s crucial to handle this situation gracefully without waiting indefinitely. In such cases, we might see bursts of 600 or more incoming requests that do not overlap with each other. To ensure a smooth operation, retries are essential.

For these instances, ElixirRetry—an amazingly powerful library for handling retries—comes into play. It provides built-in retry logic with backoff strategies, ensuring that the system can handle transient issues without being left hanging or overloaded.

Here's where the bespoke service I developed Ping Pong that comes into play for balancing loads and ensuring no request drops due to rate limit constraints, with its source code available at https://github.com/ahsandar/ping_pong.

The rate limiter GenServer code is as below


defmodule PingPong.RateLimiter do
  use GenServer
  require Logger

  alias PingPong.Utility, as: Utility
  @rate_limit_window 60_000
  @safe_limit 1

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

  def init(config) do
    Cachex.put(:ping_pong, Utility.cachex_counter(), 0)

    {:ok,
     %{
       name: config[:name],
       start_time: Utility.datetime_iso8601(),
       rate_limit: String.to_integer(config[:rate_limit] || "60")
     }}
  end

  def handle_call(:control_rate_limit, _from, state) do
    sleep_time = (@rate_limit_window / state.rate_limit) |> trunc()

    {_, count} = Cachex.get(:ping_pong, Utility.cachex_counter())
    Logger.info("Count since #{state.start_time}: #{count}")

    if count > @safe_limit do
      Logger.info(
        "Ensuring #{state.name} rate limit at #{Utility.datetime_iso8601()}, waiting for #{sleep_time}"
      )

      :timer.sleep(sleep_time)
    else
      Logger.info("Request count in safe zone")
    end

    {_, count} = Cachex.decr(:ping_pong, Utility.cachex_counter())
    Logger.info("Count since #{state.start_time}: #{count}")
    {:reply, :ok, state}
  end

  def queue(name, timeout \\ :infinity) do
    GenServer.call(name, :control_rate_limit, timeout)
  end
end

Pooling AMQP TLS connections in Elixir for high throughput and low latency

Ahsan Nabi Dar — Sun, 15 Sep 2024 03:33:10 +0000

In the fast-paced, data-driven world of computing, reliable message brokers are critical for communication between services. One of the most widely adopted protocols in this domain is the Advanced Message Queuing Protocol (AMQP). As an open standard, AMQP has seen large-scale adoption across industries for its reliability, flexibility, and ability to handle both transactional and high-throughput workloads.

AMQP serves as the backbone for several popular messaging systems, most notably RabbitMQ, which has become a go-to solution for many enterprise clients. But what makes RabbitMQ shine in comparison to alternatives like Kafka, and why is it favored by a wide range of businesses, from startups to established enterprises?

The AMQP protocol is designed to ensure safe, guaranteed delivery of messages with features like message acknowledgments, routing, and delivery confirmations. Enterprises with complex architectures rely on these guarantees to keep their systems resilient. AMQP allows companies to achieve:

Transactional messaging: Ideal for financial systems or any environment where guaranteed message delivery is crucial.

Flexible message routing: Through its exchange types (direct, topic, fanout, headers), AMQP makes it easier to handle sophisticated routing logic.

Protocol independence: Since it's an open standard, AMQP works across multiple languages, platforms, and frameworks, making it adaptable to various enterprise ecosystems.

This versatility has led to AMQP being the protocol of choice for major players in industries such as finance, retail, and telecommunications

TCP based connections can be resource-intensive, and in high-concurrency environments, it is crucial to manage connections efficiently. Using connection pooling allows developers to reuse connections instead of creating new ones for every request. This reduces:

Latency: Establishing a new connection for every task introduces delays. Connection pools maintain a set of reusable connections that can be quickly allocated to new requests.

Resource Utilization: Opening too many connections can exhaust memory and CPU resources, especially in large-scale deployments. Pooling ensures that the system remains performant and avoids unnecessary overhead.

By optimizing connection management, pooling helps maintain efficiency even in high-load environments, making it better suited for applications.

Elixir, a modern programming language that runs on the BEAM VM, provides strong support for working with AMQP. Elixir brings:

Concurrency: Leveraging the BEAM’s concurrency model, Elixir developers can easily build distributed systems that communicate over AMQP, without worrying about thread management.

Robust Libraries: Libraries like AMQP and Broadway in Elixir allow developers to create pipelines for processing AMQP messages with ease. These libraries offer a higher level of abstraction, allowing developers to focus more on business logic rather than low-level messaging code.

Fault-Tolerant Systems: Since Elixir inherits the fault-tolerance properties of BEAM, developers can build resilient, highly available systems that can recover from failure seamlessly when handling message queues.

For developers building messaging systems with AMQP, Elixir provides a natural and powerful toolset for building high-performance, fault-tolerant applications that scale efficiently.

Now on to the interesting part, how to implement connection pooling in elixir for AMQP using Poolboy for publishing messages and setting up Broadway for consumption

add amqp, poolboy, broadway and broadway_rabbitmq to your dependencies.

GenServer to maintain connections and channel and create queues upon initialization

reconnection logic used from Conduit AMQP

defmodule Maverick.Amqp.Broker do
  @moduledoc """
  Maverick.Amqp.Broker
  """

  require Logger

  use GenServer

  @exchange "topgun"

  def host(),
    do: Application.fetch_env!(:maverick, :amqp_host)

  def vhost(),
    do: Application.fetch_env!(:maverick, :amqp_vhost)

  def port(),
    do: Application.fetch_env!(:maverick, :amqp_port)

  def username(),
    do: Application.fetch_env!(:maverick, :amqp_username)

  def password(),
    do: Application.fetch_env!(:maverick, :amqp_password)

  def exchange(), do: @exchange

  def start_link(_config) do
    GenServer.start_link(__MODULE__, [])
  end

  @impl true
  def init(_config) do
   connection()
  end

  @impl true
  def handle_call(:get_channel, _from, state) do
    {:reply, state.channel, state}
  end

  @impl true
  def handle_call({:create_queues, topics}, _from, state) do
    queues(state.channel, topics, @exchange)
  end

  def handle_info(:reconnect, _from, state) do
  case connection() do
      {:ok, %{connection: conn, channel: chan}} ->
        {:noreply, %{connection: conn, channel: chan}}
      {:error, _} ->
        # Schedule a retry with a delay
        Process.send_after(self(), :reconnect, 5000)
        {:noreply, state}
    end 
  end

  def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
      Process.send_after(self(), :reconnect, 1000)
      {:noreply, state}
  end

  @impl true
  def terminate(reason, %{connection: connection}) do
    Logger.info("AMQP Broker termintated: #{reason}")
    AMQP.Connection.close(connection)
  end

  def channel(worker) do
    GenServer.call(worker, :get_channel)
  end

  def create_queues(queues) do
    GenServer.call(__MODULE__, {:create_queues, queues})
  end

  defp connection() do
    case connect() do
    {:ok, conn}  -> 
      Process.monitor(conn.pid)
      {:ok, chan} = create_channel(conn)
      :ok = exchange(chan)
      :ok = queues(chan, Maverick.Queue.Gateway.topics())
      # Keep channel open for the GenServer lifecycle
      {:ok, %{connection: conn, channel: chan}}
    {:error, reason} -> {:error, reason}
    end

  end

  defp connect() do
    AMQP.Connection.open(
      username: username(),
      password: password(),
      virtual_host: vhost(),
      host: host(),
      port: port(),
      ssl_options: [
        verify: :verify_peer,
        customize_hostname_check: [
          match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
        ],
        # from CAStore package
        cacertfile: CAStore.file_path()
      ]
    )
  end

  defp create_channel(conn) do
    AMQP.Channel.open(conn)
  end

  defp exchange(chan, exchange \\ @exchange) do
    AMQP.Exchange.declare(chan, exchange)
  end

  defp queues(chan, topics, exchange \\ @exchange) do
    Enum.each(topics, fn topic ->
      AMQP.Queue.declare(chan, topic, durable: true)
      AMQP.Queue.bind(chan, topic, exchange, routing_key: topic)
    end)
  end
end

AMQP client implementing message publishing

defmodule Maverick.Amqp.Client do


  use Retry.Annotation
  use Appsignal.Instrumentation.Decorators

  alias Maverick.Amqp.Broker, as: AmqpBroker

  @timeout 60_000

  @decorate transaction_event()
  @retry with: exponential_backoff() |> randomize() |> expiry(10_000)
  def publish(message, topic) do
    :poolboy.transaction(
      :amqp_worker,
      fn worker ->
        AMQP.Basic.publish(AmqpBroker.channel(worker), AmqpBroker.exchange(), topic, message)
      end,
      @timeout
    )
  end
end

poolboy config to start a pool of connections

import Config

config :maverick, :amqp_worker,
  pool_size: 2,
  max_overflow: 1

 defp amqp_poolboy_config() do
    [
      {:name, {:local, :amqp_worker}},
      {:worker_module, Maverick.Amqp.Broker},
      {:size, Application.get_env(:maverick, :amqp_worker)[:pool_size]},
      {:max_overflow, Application.get_env(:maverick, :amqp_worker)[:max_overflow]}
    ]
  end

Add poolboy to the Supervisor

children = [
:poolboy.child_spec(:amqp_worker, amqp_poolboy_config())
]

 opts = [strategy: :one_for_one, name: Maverick.Supervisor]
    Supervisor.start_link(children, opts)

Broadway to consume messages

defmodule Maverick.Broadway.Amqp.Whatsapp do
  @moduledoc """
  Maverick.Broadway.Amqp.Whatsapp
  """
  use Broadway

  use Appsignal.Instrumentation.Decorators

  alias Maverick.Amqp.Broker, as: AmqpBroker
  alias Maverick.Whatsapp.Message, as: WhatsappMessage

  def producer_concurrency(),
    do: Application.fetch_env!(:maverick, :broadway_producer_concurrency)

  def processors_concurrency(),
    do: Application.fetch_env!(:maverick, :broadway_processor_concurrency)

  def group(), do: Application.fetch_env!(:maverick, :broadway_client_prefix)

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module:
          {BroadwayRabbitMQ.Producer,
           queue: Application.get_env(:maverick, :whatsapp_topic),
           on_success: :ack,
           on_failure: :reject_and_requeue_once,
           declare: [durable: true],
           connection: [
             username: AmqpBroker.username(),
             password: AmqpBroker.password(),
             virtual_host: AmqpBroker.vhost(),
             host: AmqpBroker.host(),
             port: AmqpBroker.port(),
             ssl_options: [
               verify: :verify_peer,
               customize_hostname_check: [
                 match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
               ],
               # from CAStore package
               cacertfile: CAStore.file_path()
             ]
           ],
           qos: [
             prefetch_count: 50
           ]},
        concurrency: producer_concurrency()
      ],
      processors: [
        default: [
          concurrency: processors_concurrency()
        ]
      ]
    )
  end

  @decorate transaction(:broadway_amqp_whatsapp_message)
  def handle_message(_, message, _) do
    IO.inspect(message, label: "Got Whatsapp message")
    message
  end
end

AMQP’s adoption continues to grow as more enterprises recognize the need for reliable, transactional messaging solutions. AMQP, with its foundation, shines in environments where reliability, message routing flexibility, and ease of use are critical. Resilient platform, especially when paired with languages like Elixir, which provides native support for AMQP-based systems provides stability and robustness.

For those looking to build robust, scalable messaging systems, AMQP’s advantages over Kafka in transactional and low-latency messaging make it a strong contender in the world of message brokers. As industries increasingly move toward distributed architectures, AMQP's proven reliability and performance in business critical environments ensure its continued success.

Simplifying Webhook Handling with Vector.dev: A Modern Solution for Serverless Apps

Ahsan Nabi Dar — Thu, 05 Sep 2024 15:01:50 +0000

Simplifying Webhook Handling with Vector.dev: A Modern Solution for Serverless Apps
In the world of serverless applications and third-party services, handling incoming data efficiently is critical. Webhooks are one common mechanism for delivering this data, where external systems push data to your service in real time. However, building and maintaining your own webhook service can quickly turn into a case of over-engineering, especially when you need to integrate multiple downstream or upstream systems. This is where Vector.dev comes into play.

Vector is a high-performance observability and data pipeline tool built in Rust, designed to handle large-scale data processing efficiently. It comes packed with a wide variety of sinks, allowing you to seamlessly push data to many destinations, all through configuration rather than writing custom code. With Vector, setting up webhook handling becomes a simple, yet powerful, solution for routing, transforming, and managing data from different sources.

Why Vector for Webhook Handling?
There are several reasons why Vector is a great fit for managing webhook events:

No Over-Engineering: You can avoid building a custom webhook service from scratch. With Vector’s pre-built sinks and powerful configurations, you can route incoming webhook data to a variety of destinations—queues, databases, object storage, and more—without extra coding.

Wide Range of Sinks: Vector supports a large ecosystem of sinks. Whether you're publishing events to Apache Kafka for real-time stream processing or backing up the data to Amazon S3 for long-term storage, Vector has you covered. And the best part? It’s all done through configuration.

Data Transformation with VRL: The Vector Remap Language (VRL) is a built-in language that allows you to remap, filter, and transform incoming webhook events. You can easily modify the structure of the incoming payload, apply filtering rules, and send the transformed data to the desired destination.

Multi-Destination Routing: Vector supports publishing the same event to multiple destinations simultaneously. For instance, you could send the incoming webhook event to Kafka for real-time processing and store a backup copy of the same event in S3—ensuring you have both live and persistent copies of the data.

Buffering and Retry Mechanisms: One of Vector’s most useful features is its built-in buffering and retry capabilities. This ensures that even if a sink becomes temporarily unavailable, Vector will hold onto the data and retry sending it, maintaining the reliability of your pipeline.

Scalability and Horizontal Scaling: Vector’s architecture is built for performance. It uses Rust for high efficiency and can scale horizontally to meet your application's growing demands. You can also connect multiple Vector nodes to pass data between them, allowing you to build resilient, fail-safe pipelines that handle data efficiently at any scale.

To setup webhook http_server source needs to be configured and it can have authentication added to ensure security.

[sources.webhook]
  type = "http_server"
  address = "0.0.0.0:8080"
  method = "PUT"
  path = "/webh00k"
  path_key = "webhook"
[sources.webhook.auth]
  username = "${BASIC_AUTH_USERNAME}"
  password = "${BASIC_AUTH_PASSWORD}"
[sources.webhook.decoding]
  codec = "json"
[sources.webhook.framing]
  method = "newline_delimited"

Once events are received by the source events can be routed to different sinks as separate streams using the route transform as such

[transforms.condition]
  type = "route"
  inputs = [ "webhook" ]
[transforms.condition.route]
 log_webhook = 'exists(.event) && .event == "log"'
 event_webhook = 'exists(.event) && .event == "event"'
 metric_webhook = 'exists(.event) && .event == "metric"'

The important point to remember is if you are producing your own data make sure you tag all your events well.

After which events can be sent over to any sinks of choice. ;)

Using Vector to handle webhooks simplifies what would otherwise be a complex and time-consuming task. By leveraging Vector’s pre-built sinks, VRL for transformation, and powerful features like buffering, retries, and horizontal scaling, you can build robust, fail-safe pipelines that scale with your application.

With minimal setup, Vector allows you to focus on what matters: delivering value, instead of getting bogged down by infrastructure concerns. Whether you're managing incoming data from third-party services or building a scalable serverless app, Vector.dev is a modern, efficient solution that reduces overhead and improves reliability.

Multi-Layered Caching with Decorators in Elixir: Optimizing Performance and Scalability

Ahsan Nabi Dar — Tue, 18 Jun 2024 07:30:18 +0000

Phil Karlton's famous quote aptly captures the essence of caching in computer science: "There are only two hard things in Computer Science: cache invalidation and naming things." Caching is a powerful technique to improve application performance by storing frequently accessed data in a faster, more readily accessible location. However, effective cache invalidation remains a challenge.

Caching is a difficult problem and invalidating a cache is even more difficult. Elixir, with its powerful in-memory caching options such as ETS and DETS, provides a robust solution that often eliminates the need for external caching systems like Memcached or Redis. For a deep dive, Dashbit’s blog post on why you may not need Redis with Elixir is an excellent resource. But for globally distributed, dynamically scaling serverless environments, these in-memory caches become less suitable.

However, scaling caching across multiple global regions in a serverless environment introduces new challenges. Dynamic scaling of nodes means that your in-memory cache can disappear along with the node, complicating matters.

Cachex, the most popular Elixir caching library, offers a clustered cache option but lacks support for dynamic node configuration. This limitation becomes evident in environments like Fly.io, where nodes scale dynamically and their addresses aren’t known at startup. Engaging with the Fly.io community led me to an open issue on Cachex’s GitHub regarding dynamic node configuration. This search introduced me to Nebulex, a feature-rich library supporting multiple cache stores, including Cachex and Redis.

Nebulex also supports various caching patterns out of the box. While setting up the Redis adapter for Nebulex, I encountered a blocker issue with Upstash Redis, revealing another limitation. Despite this, the exploration provided insights on constructing a layered caching solution using decorators a approach Nebulex uses.

With the limitations of Cachex in dynamic environments and the Redis adapter for Nebulex not meeting expectations, Inspired by the strengths of Cachex and Nebulex, we can create a custom layered caching solution using decorators in Elixir. This approach combines the speed of a local Cachex as L1 (short TTL) with the scalability of an external Redis as L2 (longer TTL)), reducing code clutter and improving maintainability. This setup avoids cluttering the code with numerous get, set, and delete operations by using decorators.

Advantages of Multi-Layered Caching

Faster Responses: Cachex (L1) provides microsecond latency for frequently accessed data.
Reduced External Hits: Redis (L2) serves as a fallback, significantly reducing the number of requests to external services.
Cost Efficiency: Fewer requests to Redis minimize cloud costs.
Graceful Degradation: The system can serve stale data from L1 within a threshold, ensuring continuity while fetching fresh data from L2.

Here's an outline of the approach:

Define the Decorator: Create a decorator function that accepts the actual function to be wrapped and the cache configuration options (e.g., cache type, TTLs for L1 and L2).
Check L1 Cache: Inside the decorator, first check the L1 cache (Cachex) for the requested data using the key derived from the function arguments.
Retrieve from L2 Cache: If the data is not found in L1, retrieve it from the L2 cache (Redis) using the same key.
Fetch from Source: If both L1 and L2 caches miss, call the wrapped function to fetch the data from the original source.
Cache the Result: Store the fetched data in both L1 and L2 caches with their respective TTLs.
Return the Data: Finally, return the retrieved or fetched data.
By wrapping functions with this decorator, you can transparently introduce caching without altering the core logic of your application.

Using Elixir's decorator library it as simple as writing it as such

Decorator class allowing you to decorate your functions as
@decorate lx_fetch(["key", args0])
@decorate lx_evict(["key", args0])

defmodule Maverick.Cache.Decorator do
  @moduledoc """
  Maverick.Cache.Decorator
  """

  use Decorator.Define, lx_fetch: 1, lx_evict: 1

  alias Maverick.Cache.L1, as: CacheL1
  alias Maverick.Cache.L2, as: CacheL2

  def lx_fetch(attrs, body, _context) do
    quote do
      case CacheL1.get(unquote(attrs)) do
        {:ok, nil} ->
          case CacheL2.get(unquote(attrs)) do
            {:ok, nil} ->
              unquote(body)

            {:ok, data} ->
              CacheL1.set(unquote(attrs), data)
              data

            {:error, _reason} ->
              unquote(body)
          end

        {:ok, data} ->
          data

        {:error, _reason} ->
          unquote(body)
      end
    end
  end

  def lx_evict(attrs, body, _context) do
    quote do
      CacheL1.del(unquote(attrs))
      CacheL2.del(unquote(attrs))
      unquote(body)
    end
  end
end

L1 cache

defmodule Maverick.Cache.L1 do
  @moduledoc """
  Maverick.Cache.L1
  """
  require Logger

  use Appsignal.Instrumentation.Decorators

  alias Maverick.Utility.Helper

  @cachex :maverick_cachex
  @ttl 1800

  @decorate transaction_event()
  def get(key) do
    cache_key = Helper.generate_cache_key(key)
    Logger.info("Get from L1 Cache for #{cache_key}")
    Cachex.get(@cachex, cache_key)
  end

  @decorate transaction_event()
  def set(key, value, ttl \\ @ttl) do
    cache_key = Helper.generate_cache_key(key)
    Logger.info("Set in L1 Cache for #{cache_key}")
    Cachex.put(@cachex, cache_key, value, ttl: :timer.seconds(ttl))
  end

  @decorate transaction_event()
  def del(key) do
    cache_key = Helper.generate_cache_key(key)
    Logger.info("Delete from L1 Cache for #{cache_key}")
    Cachex.del(@cachex, cache_key)
  end
end

L2 cache

defmodule Maverick.Cache.L2 do
  @moduledoc """
  Maverick.Cache.L2
  """
  require Logger

  use Appsignal.Instrumentation.Decorators

  alias Maverick.Utility.Helper

  @redis :maverick_redix
  @ttl 7_776_000

  @decorate transaction_event()
  def get(key) do
    cache_key = Helper.generate_cache_key(key)

    Logger.info("Get from L2 Cache for #{cache_key}")

    case Redix.command(@redis, ["GET", cache_key]) do
      {:ok, nil} -> {:ok, nil}
      {:ok, data} -> {:ok, :erlang.binary_to_term(data)}
      {:error, reason} -> {:error, reason}
    end
  end

  @decorate transaction_event()
  def set(key, value, ttl \\ @ttl) do
    cache_key = Helper.generate_cache_key(key)
    Logger.info("Set in L2 Cache for #{cache_key}")
    Redix.command(@redis, ["SET", cache_key, :erlang.term_to_binary(value), "EX", ttl])
  end

  @decorate transaction_event()
  def del(key) do
    cache_key = Helper.generate_cache_key(key)
    Logger.info("Delete from L2 Cache for #{cache_key}")
    Redix.command(@redis, ["DEL", cache_key])
  end

  def flush_all() do
    Redix.command(:redix, ["FLUSHALL"])
    |> case do
      {:ok, status} -> status
      {:error, reason} -> reason
    end
  end
end

Cache key generation is done as in Nebulex using erlang's phash/2

def generate_cache_key(key) do
    :erlang.phash2(key) |> Integer.to_string()
end

And my favourite version of Phil Karlton's quote is:

"There are 2 hard problems in computer science: cache invalidation, naming things, and off-by-1 errors."
-- Leon Bambrick

Brod; boss kafka in elixir

Ahsan Nabi Dar — Mon, 03 Jun 2024 15:32:32 +0000

Elixir has rapidly gained recognition as a powerful language for developing distributed systems. Its concurrency model, based on the Erlang VM (BEAM), allows developers to build highly scalable and fault-tolerant applications. One of the areas where Elixir shines is in reading messages from Kafka using Broadway to build concurrent and multi-stage data ingestion and processing pipelines. However, when it comes to producing messages to Kafka, Elixir's ecosystem seems to lack a unified focus, leading to some confusion.

Elixir's inherent support for concurrency and fault tolerance makes it an ideal choice for distributed systems. The language's lightweight process model, along with features like supervisors and the actor model, enables developers to create systems that can handle massive loads and recover gracefully from failures. This makes Elixir particularly well-suited for distributed systems, where reliability and performance are crucial.

In the Elixir ecosystem, there is a clear focus on consuming messages from Kafka. Libraries like Broadway make it easy to build sophisticated data ingestion pipelines. Broadway allows developers to define multi-stage pipelines that can process large volumes of data concurrently, leveraging Elixir's strengths in concurrency and fault tolerance.

While Broadway excels at consuming messages, there's a slight terminology hiccup. The method for sending messages is called produce, which might be a bit confusing. It's important to remember that Broadway focuses on the consumer side of the Kafka equation.

However, when it comes to producing messages to Kafka, Elixir's libraries present a more fragmented landscape.

There are three primary Kafka libraries available for Elixir developers: brod, kaffe, and kafka_ex. Each of these libraries has its own strengths and use cases.

brod: An Erlang client for Kafka, brod is known for its robustness and performance. It operates seamlessly within the BEAM ecosystem, taking advantage of Erlang's mature infrastructure for building distributed systems. However, working with brod can be cumbersome and requires a fair amount of setup. Despite this, it remains a reliable and performant choice for Kafka integration.

kaffe: A wrapper around brod, kaffe simplifies the process of interacting with Kafka, particularly for those using Heroku Kafka clusters. By abstracting away some of the complexities of brod, kaffe makes it easier for developers to get started with Kafka in Elixir. It focuses on providing a more user-friendly experience while still leveraging the underlying power of brod.

kafka_ex: Currently in a state of transition, kafka_ex is undergoing significant changes that are not backward compatible. This situation can be likened to Python's shift from version 2 to version 3, where developers faced considerable breaking changes. While kafka_ex has been a popular choice, the ongoing transition means developers need to be cautious about using it in production until the changes stabilize.

Keeping in mind all the above and scouring elixirforum.com the most shared opinion was to go with brod and write a wrapper around it.

Something that lacks is over how to set it up in elixir and in particular phoenix as brod is an erlang library.

To setup brod in phoenix you need to define a supervisor process that can than be added to your phoenix supervisor like other process to be part of your supervision tree.

To setup the brod client with sasl and ssl as a supervisor it can be done as below.

defmodule Maverick.Kafka.BrodSupervisor do
  @moduledoc """
  Maverick.Kafka.BrodSupervisor
  """
  use Supervisor

  alias Maverick.Kafka, as: Kafka

  def start_link(_config) do
    Supervisor.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init(_config) do
    :ok =
      :brod.start_client(
        Kafka.hosts(),
        Kafka.brod_client(),
        ssl: true,
        ssl_options: [
          # from CAStore package
          cacertfile: CAStore.file_path(),
          verify_type: :verify_peer,
          customize_hostname_check: [
            match_fun: :public_key.pkix_verify_hostname_match_fun(:https)
          ]
        ],
        sasl: Kafka.authentication(),
        auto_start_producers: true,
        reconnect_cool_down_seconds: 10,
        default_producer_config: [
          required_acks: -1,
          partition_buffer_limit: 1024
        ]
      )

    children = []
    Supervisor.init(children, strategy: :one_for_one)
  end
end

After this you can wrap the message producer for convenience

defmodule Maverick.Kafka.Brod do
  @moduledoc """
  Maverick.Kafka.Brod
  """

  use Retry.Annotation

  @retry with: exponential_backoff() |> randomize() |> expiry(10_000)
  def produce(client, topic, partition, key, msg) do
    :brod.produce_sync(client, topic, partition, key, msg)
  end
end

I highly recommend to check out retry hex package it will save you from all those network issues as network is always dubious.

Before I close this post the thing I would like to highlight is the required_acks settings and what it means the required_acks (also known as acks) setting determines how many acknowledgements the producer requires the leader to have received before considering a request complete. This setting has a significant impact on the durability and consistency guarantees of your messages. The required_acks can be set to different values:

0: The producer does not wait for any acknowledgement from the server at all. This means that the producer will not receive any acknowledgement for the messages sent, and message loss can occur if the server fails before the message is written to disk.
1: The leader writes the record to its local log but responds without waiting for full acknowledgement from all followers. This means that the message is acknowledged as soon as the leader writes it, but before all replicas have received it.
-1 (or all): The leader waits for the full set of in-sync replicas to acknowledge the record. This is the strongest guarantee and means that the message is considered committed only when all in-sync replicas have acknowledged it.

In the context of the brod library, setting required_acks: -1 ensures that:

The producer waits for acknowledgements from all in-sync replicas before considering the message successfully sent.
This provides the highest level of durability since the message will be available even if the leader broker fails after the message is acknowledged.

I hope this makes it simple for people looking to work with Kafka using Elixir.

Deploy Ollama with s6-overlay to serve and pull in one shot

Ahsan Nabi Dar — Wed, 29 May 2024 17:27:44 +0000

Ollama brings the power of Large Language Models (LLMs) directly to your local machine. It removes the complexity of cloud-based solutions by offering a user-friendly framework for running these powerful models.

Ollama is a robust platform designed to simplify the process of running machine learning models locally. It offers an intuitive interface that allows users to efficiently manage and deploy models without the need for extensive technical knowledge. By streamlining the setup and execution processes, Ollama makes it accessible for developers to harness the power of advanced models directly on their local machines, promoting ease of use and faster iterations in development cycles.

However, Ollama does come with a notable limitation when it comes to containerized deployments. To download and manage models, Ollama must be actively running and serving before the models can be accessed. This requirement complicates the deployment process within containers, as it necessitates additional steps to ensure the service is up and operational before any model interactions can occur. Consequently, this adds complexity to Continuous Integration (CI) and Continuous Deployment (CD) pipelines, potentially hindering seamless automation and scaling efforts.

On Ollama's docker hub it has clear instructions over how to run Ollama requiring 2 steps. In the 1st step you need to have Ollama running before you can download the model to have it ready for prompting.

docker run -d --gpus=all -v ollama:/root/.ollama -p 11434:11434 --name ollama ollama/ollama

docker exec -it ollama ollama run llama3

On their Discord there is a help query about how to do this in one shot with a solution which is good but not something I would put in production to lack of orchestration and supervision of processes. Its on github as autollama and I recommend to check it out to learn some new tricks.

This is where I leveraged my past experience of using s6-overlay to setup serve and pull in a single container with serve as a longrun and pull as a oneshot dependent on serve to be up and running.

The directory structure for it as below

It runs flawlessly with pull running well supervised and orchestrated for it to complete and even when the download gets hammered due to internet speeds it keeps the process going without a glitch.

Currently there is a known issue in s6-overlay for service wait time which initially caused the oneshot to timeout. Had to S6_CMD_WAIT_FOR_SERVICES_MAXTIME=0 to disable it for the model download to not fail.

It is alive, at this point I was just super happy how smoothly it came up

On following run pull only gets the diff if any without the need to download the whole model again.

And Ollama has an api that you can prompt and its a charm to play around with.

With serve and pull in a single container to be served along your application it simplifies not only your deployments but also your CI to test it without overly complicating things by hacking scripts.

I have put the repo on github as ollama-s6 for anyone looking to productionize their ollama deplyoments.

Supercharge Your Ecto Queries over Postgres JSONB with Flop: Filtering, Sorting, and Pagination

Ahsan Nabi Dar — Thu, 28 Mar 2024 10:13:33 +0000

Elixir's Ecto library is fantastic for interacting with your database. But when it comes to building dynamic user interfaces with features like filtering, sorting, and pagination, things can get a bit cumbersome.

Enter Flop, a powerful Elixir library designed to simplify these tasks. Flop seamlessly integrates with Ecto, allowing you to add robust filtering, sorting, and pagination capabilities to your queries with minimal effort.

Flop allows you to define custom filters for specific scenarios, providing ultimate flexibility in how you handle user input.Filter and sort data across related tables using join fields, ensuring a consistent user experience.

Postgres is master DB and its support for JSONB columns offer a flexible way to store structured data within your database, but querying them can present some challenges compared to traditional columns. Ranging from Limited Indexing, Query Complexity and Optimizer Difficulties to name a few.

Data can be extracted utilizing operators like ->>, @>, and others designed for querying JSON data. These operators allow you to navigate the JSON structure and extract specific values for filtering or sorting.

Translating such operators for dynamic queries is not straightforward. Anyone who has ever added filter would understand the work required to implement the filters, operators and avoid SQL injection attacks.

Flop works seamlessly over traditional database columns out of the box and also allows great features to work with joining fields across tables by defining filters with ease using schema configurations. Something that is missing in the docs is how to use flop query JSONB column fields.

So I reached out to the hex package author on Elixir forum and he suggested it could be done using join fields or custom fields. I opted for custom fields over join fields as it didn't made sense to me to do a self join to find a value in this case.

My initial implementation didn't work as expected and I post on elixir forum and the author provided valuable guidance over debugging the sql being generated.

Few things that were to be done differently than expected for custom fields in case of querying jsonb. Below is how you can define custom fields for a jsonb column named metadata with active and owner fields.

 adapter_opts: [
      custom_fields: [
        metadata_active: [
          filter: {MetadataFilter, :metadata, []},
          ecto_type: :boolean,
          operators: [:==]
        ],
        metadata_owner: [
          filter: {MetadataFilter, :metadata, []},
          ecto_type: :string,
          operators: [:==]
        ]
      ]
    ]

Filtering functionality can be defined as below

defmodule Maverick.Utility.MetadataFilter do
  @moduledoc """
   Maverick.Utility.MetadataFilter
  """

  import Ecto.Query

  def metadata(query, %Flop.Filter{field: name, value: value, op: op} = _flop_filter, _) do
    metadata_value = value(name, value)

    expr = dynamic_expr(name)

    case metadata_value do
      {:ok, query_value} ->
        conditions =
          case op do
            :== -> dynamic([r], ^expr == ^query_value)
            :!= -> dynamic([r], ^expr != ^query_value)
            :> -> dynamic([r], ^expr > ^query_value)
            :< -> dynamic([r], ^expr < ^query_value)
            :>= -> dynamic([r], ^expr >= ^query_value)
            :<= -> dynamic([r], ^expr <= ^query_value)
          end

        where(query, ^conditions)

      :error ->
        IO.inspect("Error casting value #{value} for #{name}")
        query
    end
  end

  def field(:metadata_active), do: :active
  def field(:metadata_owner), do: :owner

  def value(:metadata_active, value), do: Ecto.Type.cast(:boolean, value)
  def value(:metadata_owner, value), do: Ecto.Type.cast(:string, value)

  def dynamic_expr(:metadata_active) do
    dynamic(
      [r],
      fragment(
        "(?->>'active')::boolean",
        field(r, :metadata)
      )
    )
  end

  def dynamic_expr(:metadata_owner) do
    dynamic(
      [r],
      fragment(
        "(?->>'owner')",
        field(r, :metadata)
      )
    )
  end
end

The record can then be filtered via API by just adding the fields as params

<url>?filters[0][field]=name&filters[0][value]=Expert&filters[1][field]=mobile_no&filters[1][value]=88827271111&filters[2][field]=metadata_active&filters[2][value]=true

The part that was to be figured out was how dynamic fragments were getting generated as using the field name would generate SQL with it as a column rather than a filed within the metadata column.

Flop is a game-changer for building dynamic and user-friendly Elixir applications. Its seamless integration with Ecto and its rich feature set make it an essential tool for streamlining data manipulation and enhancing the user experience. So, if you're looking to simplify filtering, sorting, and pagination in your Elixir projects, Flop is definitely worth exploring.

It was great working with the package author to implement and debug such functionality. Elixir community is amazing and probably the best one out there.

Also I would like to thank ChatGPT and Gemini for outputting garbage and hallucinating solutions wasting hours of valuable time that were only recovered by the help of the author and the elixir community.

Using Plug in Elixir Phoenix to transfer custom request header or params value to HTTP Headers

Ahsan Nabi Dar — Tue, 23 Jan 2024 01:07:37 +0000

Elixir's web framework Phoenix makes use of Plugs which are composable module layers through which the HTTP request passes through allowing the incoming request to be manipulated as it is processed. As per Phoenix docs

Plug lives at the heart of Phoenix's HTTP layer, and Phoenix puts Plug front and center. We interact with plugs at every step of the request life-cycle, and the core Phoenix components like endpoints, routers, and controllers are all just plugs internally

Recently building an API (yes Phoenix is now my go to choice for web dev :) ) I decided to use simple_token_authentication hex package instead of hand-rolling out an API Token checker. It has some nice features where it adds service name to the log metadata for the matching token and uses erlang's persistent_term to make some performance gains. Also you should never hand-roll out Auth(n/z) by yourself to avoid missing out on edge cases or introducing loopholes.

The only shortcoming I faced with simple_token_authentication was that it requires that the API token should be sent in Authorization header for e.g Authorization: <API TOKEN>. Now this doesn't seem to be a problem when you are integrating your API with other services that you own (having API keys in mobile and web app are just useless for protection by exposing them publicly). This can be done as intended other than in case the Authorization header is not available or restricted for use and you would need to use a custom header to pass the value. Another case and more prominent one is when integrating your API with 3rd party service or webhook where you can't pass a value in a HTTP header and have to append it to your URL.

So in order to pick your API token from HTTP header X-API-TOKEN or param x_api_token. You can add a plug that sits before SimpleTokenAuthentication Plug in the pipeline and it would transfer the value over to Authorization header.

defmodule MaverickWeb.Plugs.TransferApiTokenToAuthorization do
  @moduledoc """
  The TransferApiTokenToAuthorization Plug.
  """

  import Plug.Conn

  @api_key_header "x-api-key"
  @api_key_param "x_api_key"
  def init(opts), do: opts

  def call(conn, _opts) do
    token = get_token_from_header(conn) || get_token_from_params(conn) || ""
    conn = put_req_header(conn, "authorization", token)
    conn
  end

  defp get_token_from_header(conn) do
    case get_req_header(conn, @api_key_header) do
      [val | _] -> val
      _ -> nil
    end
  end

  defp get_token_from_params(conn), do: Map.get(conn.query_params, @api_key_param)

end

You can use this to transfer values from any intended header or param to any header in your request.

Banner Image : https://blog.logrocket.com/wp-content/uploads/2022/10/phoenix-plugs-web-app-functions.png