Running large language model workloads in production is fraught with instability. Unlike typical stateless services, they are often long-running, resource-intensive, and susceptible to a unique class of failures, from upstream API timeouts to malformed outputs. A simple Kubernetes Deployment resource quickly reveals its shortcomings. When a text processing worker consuming from a message queue fails, we don’t just need it to restart; we need an intelligent control plane that understands the application’s entire state, manages its configuration declaratively, and ensures the whole pipeline remains resilient.
The core pain point was a high-volume stream of user feedback that needed near-real-time emotional analysis. Our initial prototype was a collection of Python scripts and a RabbitMQ queue, managed by hand. This was fragile. Scaling required manual adjustments, configuration changes were error-prone, and a single worker crash could halt a segment of the processing. We needed to graduate from a collection of parts to a cohesive, self-healing system.
The concept was to build a custom Kubernetes Operator. Instead of managing Pods and Deployments directly, our platform team would manage a higher-level custom resource, something like an EmotionAnalysisPipeline. This resource would describe the desired state: how many workers, which AI model to use, and how to connect to the message queue. The Operator, our custom controller, would be the tireless robot sysadmin, working 24/7 to make the reality inside the cluster match that desired state. This is the essence of the declarative API pattern and a cornerstone of Cloud Native operations.
Technology Selection and Trade-offs
The choice of a Kubernetes Operator pattern over a simpler Helm chart or a standard Deployment with a Horizontal Pod Autoscaler (HPA) was deliberate. An HPA scales on CPU or memory, but our bottleneck wasn’t raw compute; it was the number of pending messages in a queue. While KEDA (Kubernetes Event-driven Autoscaling) could solve that specific problem, an Operator gives us a framework for much more complex logic. We could, for instance, have the Operator perform a graceful shutdown sequence on a worker, ensuring it finishes its current task before terminating, or even have it provision and configure the RabbitMQ queues themselves. We’re building a platform component, not just deploying an app. For scaffolding, kubebuilder is the industry standard, providing a solid foundation for the Custom Resource Definition (CRD) and controller logic in Go.
For the message queue, RabbitMQ was chosen for its mature support for dead-lettering and flexible routing topologies, which are critical for handling processing failures. A message that repeatedly fails analysis (a “poison pill”) can’t be allowed to block the queue. It must be shunted to a Dead Letter Queue (DLQ) for later inspection. This is a non-negotiable requirement for a production-grade system.
Finally, LangChain provides the abstraction layer for the core logic inside the worker pods. Directly integrating with a model provider’s API would hardcode our workers to a specific service. LangChain lets us define a “chain”—a sequence of prompt templating, model invocation, and output parsing—that can be easily re-targeted to different models (from OpenAI’s GPT series to a locally hosted Llama model via Ollama) with minimal code changes. This architectural seam is vital for cost management and avoiding vendor lock-in.
Defining the Contract: The EmotionAnalysisPipeline CRD
Everything starts with the API. Our CRD defines the schema for the EmotionAnalysisPipeline custom resource. This is the declarative interface our users will interact with. A practical CRD needs to capture the essential knobs for the pipeline.
Here’s the Go definition using kubebuilder markers, which generate the corresponding YAML CRD.
api/v1/emotionanalysispipeline_types.go:
package v1
import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// EmotionAnalysisPipelineSpec defines the desired state of EmotionAnalysisPipeline
type EmotionAnalysisPipelineSpec struct {
// Replicas is the number of worker pods to run.
// +kubebuilder:validation:Minimum=0
Replicas *int32 `json:"replicas"`
// WorkerImage defines the container image for the analysis worker.
WorkerImage string `json:"workerImage"`
// QueueConfig contains the configuration for connecting to RabbitMQ.
QueueConfig QueueConfigSpec `json:"queueConfig"`
// ModelConfig specifies which language model to use for analysis.
ModelConfig ModelConfigSpec `json:"modelConfig"`
}
// QueueConfigSpec defines the RabbitMQ connection details.
type QueueConfigSpec struct {
// AMQP URI for the RabbitMQ connection.
AMQP_URI string `json:"amqpURI"`
// Name of the primary queue to consume from.
QueueName string `json:"queueName"`
// Name of the dead-letter queue for failed messages.
DeadLetterQueueName string `json:"deadLetterQueueName"`
}
// ModelConfigSpec defines the model configuration for LangChain.
type ModelConfigSpec struct {
// Provider specifies the model provider (e.g., "Ollama", "OpenAI").
Provider string `json:"provider"`
// ModelName is the specific model to use (e.g., "llama3", "gpt-4-turbo").
ModelName string `json:"modelName"`
// BaseURL for the model API, especially for local providers like Ollama.
BaseURL string `json:"baseURL,omitempty"`
}
// EmotionAnalysisPipelineStatus defines the observed state of EmotionAnalysisPipeline
type EmotionAnalysisPipelineStatus struct {
// ReadyReplicas is the number of worker pods that are ready.
ReadyReplicas int32 `json:"readyReplicas"`
// Conditions represent the latest available observations of the pipeline's state.
Conditions []metav1.Condition `json:"conditions,omitempty"`
}
//+kubebuilder:object:root=true
//+kubebuilder:subresource:status
//+kubebuilder:printcolumn:name="Desired",type="integer",JSONPath=".spec.replicas"
//+kubebuilder:printcolumn:name="Ready",type="integer",JSONPath=".status.readyReplicas"
//+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.conditions[?(@.type=='Ready')].reason"
//+kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
// EmotionAnalysisPipeline is the Schema for the emotionanalysispipelines API
type EmotionAnalysisPipeline struct {
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty"`
Spec EmotionAnalysisPipelineSpec `json:"spec,omitempty"`
Status EmotionAnalysisPipelineStatus `json:"status,omitempty"`
}
//+kubebuilder:object:root=true
// EmotionAnalysisPipelineList contains a list of EmotionAnalysisPipeline
type EmotionAnalysisPipelineList struct {
metav1.TypeMeta `json:",inline"`
metav1.ListMeta `json:"metadata,omitempty"`
Items []EmotionAnalysisPipeline `json:"items"`
}
func init() {
SchemeBuilder.Register(&EmotionAnalysisPipeline{}, &EmotionAnalysisPipelineList{})
}This CRD is the blueprint. It allows a user to create a resource like this:
config/samples/example_pipeline.yaml:
apiVersion: myapp.my.domain/v1
kind: EmotionAnalysisPipeline
metadata:
name: user-feedback-pipeline
spec:
replicas: 3
workerImage: "my-registry/emotion-worker:0.1.0"
queueConfig:
amqpURI: "amqp://guest:[email protected]:5672/"
queueName: "feedback-analysis-queue"
deadLetterQueueName: "feedback-analysis-dlq"
modelConfig:
provider: "Ollama"
modelName: "llama3"
baseURL: "http://ollama.default.svc.cluster.local:11434"The Workhorse: The Python Worker
The worker pod contains the actual analysis logic. It’s designed to be simple and stateless. Its only job is to connect to RabbitMQ, pull a message, process it using LangChain, and acknowledge it. All the complexity of scaling and lifecycle is offloaded to the Operator.
A robust worker must handle failures gracefully. A common mistake is to simply catch an exception and re-queue the message. If the message itself is the problem, this creates an infinite processing loop that poisons the queue. The correct pattern is to use a Dead Letter Queue.
worker/main.py:
import os
import pika
import json
import logging
import time
from langchain_community.chat_models import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
# --- Configuration ---
AMQP_URI = os.getenv("AMQP_URI")
QUEUE_NAME = os.getenv("QUEUE_NAME")
DLQ_NAME = os.getenv("DLQ_NAME")
MODEL_PROVIDER = os.getenv("MODEL_PROVIDER", "Ollama")
MODEL_NAME = os.getenv("MODEL_NAME", "llama3")
MODEL_BASE_URL = os.getenv("MODEL_BASE_URL")
# --- Logging Setup ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# --- LangChain Setup ---
# Define the desired JSON output structure using Pydantic.
# This makes the LLM's output reliable and machine-readable.
class EmotionAnalysis(BaseModel):
primary_emotion: str = Field(description="The dominant emotion detected in the text, from a list of: Joy, Sadness, Anger, Fear, Surprise, Disgust, Neutral.")
sentiment: str = Field(description="The overall sentiment, one of: Positive, Negative, Neutral.")
confidence_score: float = Field(description="A confidence score from 0.0 to 1.0 for the primary emotion detection.")
key_phrases: list[str] = Field(description="A list of key phrases from the text that indicate the primary emotion.")
# Set up the parser with the Pydantic model.
parser = JsonOutputParser(pydantic_object=EmotionAnalysis)
# Create a prompt template that instructs the model how to behave
# and specifies the required JSON format.
prompt = ChatPromptTemplate.from_messages([
("system", "You are an expert in textual emotional analysis. Analyze the user's text and respond ONLY with a JSON object matching the following schema: {format_instructions}"),
("human", "{text}"),
])
# Initialize the model. This part is now pluggable based on environment variables.
if MODEL_PROVIDER.lower() == "ollama":
llm = ChatOllama(model=MODEL_NAME, base_url=MODEL_BASE_URL, format="json")
else:
# In a real project, you would add logic for other providers like OpenAI, Anthropic, etc.
raise ValueError(f"Unsupported model provider: {MODEL_PROVIDER}")
# The complete analysis chain.
chain = prompt | llm | parser
def process_message(body):
"""
The core logic for processing a single message.
"""
try:
data = json.loads(body)
text_to_analyze = data.get("text")
if not text_to_analyze:
logging.error("Message is missing 'text' field.")
return False # Indicates a fatal error with the message
logging.info(f"Analyzing text: '{text_to_analyze[:50]}...'")
# Invoke the LangChain processing chain
result = chain.invoke({"text": text_to_analyze, "format_instructions": parser.get_format_instructions()})
logging.info(f"Analysis complete: {result}")
# In a real application, you would publish this result to another queue or database.
return True # Indicates successful processing
except json.JSONDecodeError:
logging.exception("Failed to decode message body as JSON.")
return False # This is a poison pill, don't retry.
except Exception:
# This catches errors from the LLM (e.g., API unavailable, malformed output).
logging.exception("An unexpected error occurred during analysis.")
return False # Assume transient, could be retried, but for robustness we send to DLQ.
def main():
"""
Main consumer loop. Connects to RabbitMQ and processes messages.
Includes connection retries and graceful error handling.
"""
while True:
try:
connection = pika.BlockingConnection(pika.URLParameters(AMQP_URI))
channel = connection.channel()
# Ensure queues exist. This is idempotent.
# Main queue is configured to send failed messages to the dead-letter exchange.
dlx_exchange = f'{DLQ_NAME}-exchange'
channel.exchange_declare(exchange=dlx_exchange, exchange_type='direct')
channel.queue_declare(queue=DLQ_NAME, durable=True)
channel.queue_bind(exchange=dlx_exchange, queue=DLQ_NAME, routing_key=DLQ_NAME)
channel.queue_declare(
queue=QUEUE_NAME,
durable=True,
arguments={
'x-dead-letter-exchange': dlx_exchange,
'x-dead-letter-routing-key': DLQ_NAME
}
)
logging.info(f"Connected to RabbitMQ. Waiting for messages in queue '{QUEUE_NAME}'.")
def callback(ch, method, properties, body):
if process_message(body):
# Acknowledge the message if processing was successful.
ch.basic_ack(delivery_tag=method.delivery_tag)
else:
# Reject the message. Because of the queue's dead-letter config,
# it will be sent to the DLQ instead of being re-queued.
logging.warning(f"Rejecting message and sending to DLQ: {DLQ_NAME}")
ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
# Set prefetch_count=1 to ensure a worker only has one message in flight.
# This prevents a busy worker from hoarding messages while other workers are idle.
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue=QUEUE_NAME, on_message_callback=callback)
channel.start_consuming()
except pika.exceptions.AMQPConnectionError:
logging.error("Connection to RabbitMQ failed. Retrying in 5 seconds...")
time.sleep(5)
except Exception as e:
logging.critical(f"A critical error occurred: {e}. Restarting consumer loop...")
time.sleep(10)
if __name__ == "__main__":
main()The accompanying Dockerfile is standard for a Python application, ensuring dependencies are installed and the script is executed.
worker/Dockerfile:
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
CMD ["python", "main.py"]The Brains of the Operation: The Controller’s Reconcile Loop
The heart of the Operator is the Reconcile function. This function is triggered whenever an EmotionAnalysisPipeline resource is created, updated, or deleted, or when a secondary resource it manages (like the Deployment) changes. The Reconciler’s job is to observe the current state of the cluster and take action to drive it towards the desired state defined in the CR.
The logic must be idempotent. Running the Reconcile function ten times with the same input should result in the same state as running it once. It doesn’t issue commands like “create this”; it asserts “this should exist”.
internal/controller/emotionanalysispipeline_controller.go:
package controller
import (
"context"
"fmt"
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/log"
myappv1 "github.com/my-org/emotion-operator/api/v1"
)
// EmotionAnalysisPipelineReconciler reconciles a EmotionAnalysisPipeline object
type EmotionAnalysisPipelineReconciler struct {
client.Client
Scheme *runtime.Scheme
}
//+kubebuilder:rbac:groups=myapp.my.domain,resources=emotionanalysispipelines,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=myapp.my.domain,resources=emotionanalysispipelines/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=myapp.my.domain,resources=emotionanalysispipelines/finalizers,verbs=update
//+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
func (r *EmotionAnalysisPipelineReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
logger := log.FromContext(ctx)
// 1. Fetch the EmotionAnalysisPipeline instance
pipeline := &myappv1.EmotionAnalysisPipeline{}
err := r.Get(ctx, req.NamespacedName, pipeline)
if err != nil {
if errors.IsNotFound(err) {
logger.Info("EmotionAnalysisPipeline resource not found. Ignoring since object must be deleted.")
return ctrl.Result{}, nil
}
logger.Error(err, "Failed to get EmotionAnalysisPipeline")
return ctrl.Result{}, err
}
// 2. Reconcile the child Deployment
found := &appsv1.Deployment{}
err = r.Get(ctx, types.NamespacedName{Name: pipeline.Name, Namespace: pipeline.Namespace}, found)
// If the Deployment does not exist, create it.
if err != nil && errors.IsNotFound(err) {
dep := r.deploymentForPipeline(pipeline)
logger.Info("Creating a new Deployment", "Deployment.Namespace", dep.Namespace, "Deployment.Name", dep.Name)
err = r.Create(ctx, dep)
if err != nil {
logger.Error(err, "Failed to create new Deployment", "Deployment.Namespace", dep.Namespace, "Deployment.Name", dep.Name)
return ctrl.Result{}, err
}
// Deployment created successfully - return and requeue to check status later.
return ctrl.Result{Requeue: true}, nil
} else if err != nil {
logger.Error(err, "Failed to get Deployment")
return ctrl.Result{}, err
}
// If the Deployment exists, ensure its spec matches the pipeline's spec.
// A real-world project would do a deep comparison here. For this example, we focus on replicas and image.
desiredReplicas := *pipeline.Spec.Replicas
if *found.Spec.Replicas != desiredReplicas {
logger.Info("Replicas mismatch", "current", *found.Spec.Replicas, "desired", desiredReplicas)
found.Spec.Replicas = &desiredReplicas
err = r.Update(ctx, found)
if err != nil {
logger.Error(err, "Failed to update Deployment replicas")
return ctrl.Result{}, err
}
}
desiredImage := pipeline.Spec.WorkerImage
if found.Spec.Template.Spec.Containers[0].Image != desiredImage {
logger.Info("Worker image mismatch", "current", found.Spec.Template.Spec.Containers[0].Image, "desired", desiredImage)
found.Spec.Template.Spec.Containers[0].Image = desiredImage
err = r.Update(ctx, found)
if err != nil {
logger.Error(err, "Failed to update Deployment image")
return ctrl.Result{}, err
}
}
// 3. Update the EmotionAnalysisPipeline status
// Reflect the state of the child Deployment in the parent's status field.
pipeline.Status.ReadyReplicas = found.Status.ReadyReplicas
err = r.Status().Update(ctx, pipeline)
if err != nil {
logger.Error(err, "Failed to update EmotionAnalysisPipeline status")
return ctrl.Result{}, err
}
return ctrl.Result{}, nil
}
// deploymentForPipeline returns a Deployment object for the given pipeline
func (r *EmotionAnalysisPipelineReconciler) deploymentForPipeline(p *myappv1.EmotionAnalysisPipeline) *appsv1.Deployment {
labels := map[string]string{"app": "emotion-worker", "pipeline": p.Name}
replicas := p.Spec.Replicas
dep := &appsv1.Deployment{
ObjectMeta: metav1.ObjectMeta{
Name: p.Name,
Namespace: p.Namespace,
},
Spec: appsv1.DeploymentSpec{
Replicas: replicas,
Selector: &metav1.LabelSelector{
MatchLabels: labels,
},
Template: corev1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: labels,
},
Spec: corev1.PodSpec{
Containers: []corev1.Container{{
Image: p.Spec.WorkerImage,
Name: "emotion-worker",
Env: []corev1.EnvVar{
{Name: "AMQP_URI", Value: p.Spec.QueueConfig.AMQP_URI},
{Name: "QUEUE_NAME", Value: p.Spec.QueueConfig.QueueName},
{Name: "DLQ_NAME", Value: p.Spec.QueueConfig.DeadLetterQueueName},
{Name: "MODEL_PROVIDER", Value: p.Spec.ModelConfig.Provider},
{Name: "MODEL_NAME", Value: p.Spec.ModelConfig.ModelName},
{Name: "MODEL_BASE_URL", Value: p.Spec.ModelConfig.BaseURL},
},
}},
},
},
},
}
// Set the pipeline instance as the owner and controller
// This ensures that if the EmotionAnalysisPipeline is deleted, the Deployment is garbage collected.
ctrl.SetControllerReference(p, dep, r.Scheme)
return dep
}
// SetupWithManager sets up the controller with the Manager.
func (r *EmotionAnalysisPipelineReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&myappv1.EmotionAnalysisPipeline{}).
Owns(&appsv1.Deployment{}). // This tells the controller to watch for changes to Deployments it owns.
Complete(r)
}This controller logic is the core of the system’s resilience. If a user manually deletes the worker Deployment, the Reconcile loop will be triggered, see that the Deployment is missing, and immediately re-create it. If a user edits the EmotionAnalysisPipeline YAML and changes the replica count from 3 to 5, the Reconciler will see the mismatch and scale up the Deployment. The system is always being driven back to its desired state.
Lingering Issues and Future Iterations
This implementation provides a solid, self-healing foundation, but in a real-world project, it’s just the starting point. The current replica count is static. The most critical next step would be to implement true autoscaling by having the Operator monitor RabbitMQ queue depth. This would involve querying the RabbitMQ Management API, comparing the message count against a threshold defined in the CRD, and adjusting the spec.replicas field on the EmotionAnalysisPipeline resource itself. This turns the Operator into a fully-fledged autoscaler tailored to its specific domain.
Furthermore, observability is rudimentary. Both the worker pods and the Operator itself should expose detailed Prometheus metrics. The workers could report processing latency, success/failure rates, and token usage, while the Operator could report reconciliation loop duration and errors. This data is essential for understanding system performance and cost.
Finally, the current model configuration is passed directly in the spec. A more sophisticated architecture would involve an integration with a model registry. The CRD might specify a model name and version, and the Operator would be responsible for configuring the worker to pull the correct model artifacts from a central store, enabling better governance and traceability for the AI components of the system. The existing architecture, however, provides the correct control plane to build these more advanced features upon.
