Key Takeaways
- Content-addressable storage (CAS) using cryptographic hashes prevents silent model corruption and guarantees reproducible rollbacks to known good states.
- Atomic filesystem operations, specifically POSIX `os.replace()`, enable sub-millisecond hot-swaps of model binaries on disk, eliminating race conditions during deployment.
- A Redis-backed dynamic routing table provides a resilient and low-latency mechanism for traffic splitting, crucial for canary deployments and A/B testing.
- Production-grade health-check mechanisms, implemented as decorators or middleware, can detect model degradation (e.g., NaN outputs) and automatically trigger instant traffic reallocation or fallbacks.
- Decoupling model deployment from heavy MLOps frameworks can significantly reduce inference latency and remove single points of failure.
The Problem
The challenge isn't just about training a good model; it's about getting that model into the hands of users, reliably and instantly. Many existing MLOps platforms offer model registries and deployment features, but often at the cost of added complexity, network overhead, and potential single points of failure. When you're serving millions of requests per second, even a 50ms latency introduced by an external model registry lookup or a few seconds of downtime during a model swap can translate to significant revenue loss or degraded user experience. Our goal is to serve a real-time text classification model (categorizing Netflix Tech Blog posts) with zero interruption, regardless of how frequently we update our underlying model. We need a system that feels like a simple `git push` but carries the reliability of a battle-hardened production pipeline.Data and Sources
For this demonstration, we'll be classifying the titles of recent posts from the Netflix Tech Blog. This provides a live, continuously updated stream of real-world text data, simulating the kind of dynamic input an ML system might encounter. We'll fetch this data using the `feedparser` library. * **Netflix Tech Blog RSS Feed:** `https://medium.com/feed/netflix-techblog` * **Python `feedparser` documentation:** [https://pypi.org/project/feedparser/](https://pypi.org/project/feedparser/) * **Scikit-learn documentation:** [https://scikit-learn.org/stable/](https://scikit-learn.org/stable/) * **`joblib` for model serialization:** [https://joblib.readthedocs.io/en/latest/](https://joblib.readthedocs.io/en/latest/) * **Redis official website:** [https://redis.io/](https://redis.io/) * **Python `redis-py` client:** [https://redis-py.readthedocs.io/en/stable/](https://redis-py.readthedocs.io/en/stable/) * **POSIX `os.replace()` documentation:** [https://docs.python.org/3/library/os.html#os.replace](https://docs.python.org/3/library/os.html#os.replace) Data accessed on 2023-10-27.Step 1 — Content-Addressable Model Packaging
The first challenge is ensuring that every model artifact is immutable and uniquely identifiable. If we just overwrite `model.joblib`, we lose the ability to reliably roll back or audit which exact version is running. Content-addressable storage (CAS) solves this by using a cryptographic hash of the model's contents as its identifier. This guarantees that if even a single byte changes, the hash changes, giving us a new, distinct version. We achieve this by serializing our `scikit-learn` pipeline (vectorizer and classifier) using `joblib`. After serialization, we compute its SHA-256 hash. This hash then becomes the name of the directory where the model and its metadata are stored. This structure means every deployed model version has a unique, self-verifying path.import joblib
import hashlib
import os
import shutil
from pathlib import Path
def package_model(model_pipeline, version_name="model", base_dir="models"):
"""Serializes a model pipeline and stores it in a content-addressable directory."""
temp_path = Path(f"{base_dir}/temp_model.joblib")
Path(base_dir).mkdir(parents=True, exist_ok=True)
# Serialize to a temporary file
joblib.dump(model_pipeline, temp_path)
# Compute SHA-256 hash of the serialized file
hasher = hashlib.sha256()
with open(temp_path, 'rb') as f:
while True:
chunk = f.read(4096)
if not chunk:
break
hasher.update(chunk)
model_hash = hasher.hexdigest()
# Create the content-addressable directory
model_dir = Path(f"{base_dir}/{model_hash}")
model_dir.mkdir(parents=True, exist_ok=True)
# Move the serialized model into its final hash-named directory
shutil.move(temp_path, model_dir / f"{version_name}.joblib")
print(f"Packaged model {version_name} to {model_dir}")
return model_hash, str(model_dir / f"{version_name}.joblib")
Step 2 — Atomic Filesystem Swapping
Once we have a new model version packaged, we need to make it available to the inference service without any service interruption. Traditional approaches might involve copying files, which can be slow for large models and introduce race conditions if the inference service tries to read a partially copied file. The solution lies in atomic filesystem operations. POSIX `os.replace()` (or `shutil.move` which uses it internally for same-filesystem moves) is key here. It renames a file or directory, and critically, if the target already exists, it replaces it. This operation is atomic at the filesystem level – it's either fully replaced or not at all, never in a half-written state. This means we can prepare a new model in a staging directory, and then instantly swap it with the "active" production model directory via a symlink, or by directly replacing the target directory if the inference service points to a fixed path. For a single inference server, directly replacing a symlink that the server points to is the most straightforward.def activate_model(model_hash, model_base_dir="models", active_link_name="current_model"):
"""
Creates or updates a symbolic link to point to the newly packaged model,
ensuring atomic activation.
"""
source_model_path = Path(f"{model_base_dir}/{model_hash}")
active_link_path = Path(f"{model_base_dir}/{active_link_name}")
if not source_model_path.exists():
raise FileNotFoundError(f"Model hash directory {source_model_path} not found.")
# Create a temporary symlink first to ensure target exists for os.replace
temp_link_path = Path(f"{active_link_path}.tmp")
if temp_link_path.exists():
os.unlink(temp_link_path)
os.symlink(source_model_path, temp_link_path)
# Atomically replace the old symlink with the new one
# This ensures active_link_path always points to a valid model
if active_link_path.exists():
os.replace(temp_link_path, active_link_path)
else:
# If no active link exists, just rename the temporary one
os.rename(temp_link_path, active_link_path)
print(f"Activated model {model_hash} via symlink {active_link_path}")
return str(active_link_path)
Step 3 — Dynamic Canary Routing via Redis
With atomic model activation, the next step is to control traffic to new versions. We want to route a precise percentage of live traffic to new model versions (canaries) without restarting the inference server. This is where Redis shines as a low-latency, shared routing table. We'll store a dictionary of model versions and their respective traffic weights in a Redis hash. For each incoming request, we generate a deterministic hash of its ID (or a similar unique identifier) and use this hash to decide which model version to route to, based on the weights. This ensures consistent routing for a given request ID while allowing dynamic adjustments to traffic distribution.import redis
import json
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class ModelRouter:
def __init__(self, redis_host='localhost', redis_port=6379, model_key_prefix="model_routes"):
self.r = redis.StrictRedis(host=redis_host, port=redis_port, decode_responses=True)
self.model_key_prefix = model_key_prefix
self.local_cache = {} # Fallback in case Redis is down
self.last_sync_time = 0
self.sync_interval = 5 # seconds
def _sync_from_redis(self, model_name):
"""Fetches routing weights from Redis, with a simple local cache."""
try:
raw_weights = self.r.hget(self.model_key_prefix, model_name)
if raw_weights:
self.local_cache[model_name] = json.loads(raw_weights)
self.last_sync_time = time.time()
logging.debug(f"Synced {model_name} from Redis: {self.local_cache[model_name]}")
elif model_name not in self.local_cache:
self.local_cache[model_name] = {} # Initialize if not in Redis
except redis.exceptions.ConnectionError:
logging.warning(f"Redis connection error for {model_name}. Using local cache.")
def set_weights(self, model_name, weights):
"""Sets the traffic weights for different model versions in Redis."""
try:
self.r.hset(self.model_key_prefix, model_name, json.dumps(weights))
self._sync_from_redis(model_name) # Update local cache immediately
logging.info(f"Set weights for {model_name}: {weights}")
except redis.exceptions.ConnectionError:
logging.error(f"Failed to set weights for {model_name} due to Redis connection error.")
def get_active_model_hash(self, model_name, request_id):
"""Determines which model version to use based on request_id and weights."""
if time.time() - self.last_sync_time > self.sync_interval or model_name not in self.local_cache:
self._sync_from_redis(model_name)
weights = self.local_cache.get(model_name, {})
if not weights:
logging.warning(f"No weights defined for {model_name}. Falling back to default.")
# Fallback: if no weights, return the first available model or a predefined default
return list(weights.keys())[0] if weights else None
# Calculate cumulative weights for routing
cumulative_weights = []
current_cumulative = 0
for model_hash, weight in weights.items():
current_cumulative += weight
cumulative_weights.append((model_hash, current_cumulative))
# Deterministic routing based on request_id
# Using hash() and modulo for simple distribution
bucket = hash(str(request_id)) % 100 # Scale to 0-99 for percentage weights
for model_hash, cumulative in cumulative_weights:
if bucket < cumulative:
return model_hash
return None # Should ideally not happen if total weight is 100
Step 4 — Automated Rollback and Circuit Breaking
Even with robust deployment, models can degrade in production. Latency spikes, unexpected `NaN` predictions, or other runtime errors require immediate action. Instead of manual intervention, we can build automated circuit breakers that detect these anomalies and instantly decrement canary traffic weights or route all traffic back to a stable production model. We'll simulate this by adding a health-check mechanism. If a model's prediction method raises an error or returns an invalid output (like `NaN`), our system will detect it and signal the `ModelRouter` to reduce its traffic share. In a real-world scenario, this would typically be integrated into a FastAPI middleware or a decorator around the `predict` function.import time
from concurrent.futures import ThreadPoolExecutor
class InferenceService:
def __init__(self, model_router, model_base_dir="models", model_name="classifier"):
self.model_router = model_router
self.model_base_dir = model_base_dir
self.model_name = model_name
self.loaded_models = {} # Cache for loaded model pipelines
def _load_model(self, model_hash):
"""Loads a specific model version from its content-addressable path."""
model_path = Path(f"{self.model_base_dir}/{model_hash}/classifier.joblib")
if not model_path.exists():
raise FileNotFoundError(f"Model file not found: {model_path}")
if model_hash not in self.loaded_models:
logging.info(f"Loading model: {model_path}")
self.loaded_models[model_hash] = joblib.load(model_path)
return self.loaded_models[model_hash]
def predict(self, request_id, input_data):
"""Routes request and performs prediction with health checks."""
model_hash = self.model_router.get_active_model_hash(self.model_name, request_id)
if not model_hash:
logging.error(f"No active model for {self.model_name} to route request {request_id}.")
return {"error": "No active model available."}
try:
model_pipeline = self._load_model(model_hash)
# Simulate a bad model version (e.g., v2) returning NaN for some requests
if "v2" in model_hash and hash(str(request_id)) % 10 < 3: # 30% chance for v2 to fail
raise ValueError("Simulated model degradation: NaN output detected!")
prediction = model_pipeline.predict(input_data)
return {"model_hash": model_hash, "prediction": prediction.tolist()}
except Exception as e:
logging.error(f"Model {model_hash} failed for request {request_id}: {e}")
self._handle_model_failure(model_hash)
return {"error": f"Prediction failed for model {model_hash}: {e}"}
def _handle_model_failure(self, failed_model_hash):
"""Reduces traffic to a failing model and attempts to fall back."""
current_weights = self.model_router.local_cache.get(self.model_name, {})
new_weights = current_weights.copy()
if failed_model_hash in new_weights and new_weights[failed_model_hash] > 0:
logging.warning(f"Reducing traffic to failing model {failed_model_hash}.")
new_weights[failed_model_hash] = max(0, new_weights[failed_model_hash] - 10) # Reduce by 10%
# Distribute the reduced weight to other models, or specifically to the stable one
other_models = [h for h in new_weights if h != failed_model_hash]
if other_models and new_weights[failed_model_hash] < current_weights[failed_model_hash]:
# Simple redistribution: give to the first "other" model
new_weights[other_models[0]] += (current_weights[failed_model_hash] - new_weights[failed_model_hash])
# Ensure total weight sums to 100 (or close) - simple normalization
total_weight = sum(new_weights.values())
if total_weight > 0:
new_weights = {k: int(v / total_weight * 100) for k, v in new_weights.items()}
self.model_router.set_weights(self.model_name, new_weights)
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import feedparser
import joblib
import hashlib
import os
import shutil
import time
import json
import logging
import redis
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
# --- Configuration ---
RSS_FEED_URL = 'https://medium.com/feed/netflix-techblog'
MODEL_BASE_DIR = 'models_deployment'
ACTIVE_MODEL_LINK_NAME = 'current_classifier'
MODEL_NAME_SLUG = 'tech_classifier' # Used for Redis routing
NUM_SIMULATED_REQUESTS = 100
REDIS_HOST = 'localhost'
REDIS_PORT = 6379
# --- Logging Setup ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# --- Helper Functions (from steps above) ---
def package_model(model_pipeline, version_tag, base_dir=MODEL_BASE_DIR):
"""
Serializes a model pipeline and stores it in a content-addressable directory.
Includes a version_tag in the filename for clarity, though hash is primary ID.
"""
temp_path = Path(f"{base_dir}/temp_model_{version_tag}.joblib")
Path(base_dir).mkdir(parents=True, exist_ok=True)
joblib.dump(model_pipeline, temp_path)
hasher = hashlib.sha256()
with open(temp_path, 'rb') as f:
while True:
chunk = f.read(4096)
if not chunk:
break
hasher.update(chunk)
model_hash = hasher.hexdigest()
model_dir = Path(f"{base_dir}/{model_hash}")
model_dir.mkdir(parents=True, exist_ok=True)
final_model_path = model_dir / f"{MODEL_NAME_SLUG}.joblib"
shutil.move(temp_path, final_model_path)
logging.info(f"Packaged model {version_tag} to {model_dir} (hash: {model_hash[:8]})")
return model_hash, str(final_model_path)
def activate_model(model_hash, model_base_dir=MODEL_BASE_DIR, active_link_name=ACTIVE_MODEL_LINK_NAME):
"""
Creates or updates a symbolic link to point to the newly packaged model,
ensuring atomic activation.
"""
source_model_path = Path(f"{model_base_dir}/{model_hash}")
active_link_path = Path(f"{model_base_dir}/{active_link_name}")
if not source_model_path.exists():
raise FileNotFoundError(f"Model hash directory {source_model_path} not found.")
temp_link_path = Path(f"{active_link_path}.tmp")
if temp_link_path.exists():
os.unlink(temp_link_path)
# Create a temporary symlink first, pointing to the new model's directory
os.symlink(source_model_path, temp_link_path, target_is_directory=True)
# Atomically replace the old symlink with the new one
# This ensures active_link_path always points to a valid model directory
try:
os.replace(temp_link_path, active_link_path)
except OSError as e:
# Handle cases where os.replace might fail for directories on some systems
# A safer cross-platform alternative for symlinks is a two-step: unlink then rename
if active_link_path.exists():
os.unlink(active_link_path)
os.rename(temp_link_path, active_link_path)
logging.warning(f"Using os.unlink then os.rename for atomic symlink swap: {e}")
logging.info(f"Activated model {model_hash[:8]} via symlink {active_link_path}")
return str(active_link_path)
class ModelRouter:
def __init__(self, redis_host=REDIS_HOST, redis_port=REDIS_PORT, model_key_prefix="model_routes"):
self.r = None
try:
self.r = redis.StrictRedis(host=redis_host, port=redis_port, decode_responses=True, socket_connect_timeout=1)
self.r.ping() # Test connection
logging.info("Successfully connected to Redis.")
except redis.exceptions.ConnectionError as e:
logging.error(f"Could not connect to Redis: {e}. Running in local-cache-only mode.")
self.r = None # Ensure r is None if connection failed
self.model_key_prefix = model_key_prefix
self.local_cache = {} # Fallback in case Redis is down
self.last_sync_time = 0
self.sync_interval = 5 # seconds
def _sync_from_redis(self, model_name):
"""Fetches routing weights from Redis, with a simple local cache."""
if not self.r: return # No Redis connection
try:
raw_weights = self.r.hget(self.model_key_prefix, model_name)
if raw_weights:
self.local_cache[model_name] = json.loads(raw_weights)
self.last_sync_time = time.time()
logging.debug(f"Synced {model_name} from Redis: {self.local_cache[model_name]}")
elif model_name not in self.local_cache:
self.local_cache[model_name] = {} # Initialize if not in Redis
except redis.exceptions.ConnectionError:
logging.warning(f"Redis connection error during sync for {model_name}. Using local cache.")
def set_weights(self, model_name, weights):
"""Sets the traffic weights for different model versions in Redis."""
if not self.r:
logging.warning(f"Redis not connected. Cannot set weights for {model_name}. Updating local cache only.")
self.local_cache[model_name] = weights
return
try:
self.r.hset(self.model_key_prefix, model_name, json.dumps(weights))
self._sync_from_redis(model_name) # Update local cache immediately
logging.info(f"Set weights for {model_name}: {weights}")
except redis.exceptions.ConnectionError:
logging.error(f"Failed to set weights for {model_name} due to Redis connection error.")
self.local_cache[model_name] = weights # Update local cache as fallback
def get_active_model_hash(self, model_name, request_id):
"""Determines which model version to use based on request_id and weights."""
if time.time() - self.last_sync_time > self.sync_interval or model_name not in self.local_cache:
self._sync_from_redis(model_name)
weights = self.local_cache.get(model_name, {})
if not weights:
# Fallback: if no weights, return the first available model in model_base_dir
# This is a very basic last-resort fallback. In production, you'd want a predefined stable model.
try:
# Find the 'current_classifier' symlink target
current_link_path = Path(MODEL_BASE_DIR) / ACTIVE_MODEL_LINK_NAME
if current_link_path.is_symlink():
resolved_path = current_link_path.resolve()
return resolved_path.name # The directory name is the hash
except FileNotFoundError:
pass # No active link found
logging.warning(f"No weights defined for {model_name} and no active symlink. Cannot route request {request_id}.")
return None
cumulative_weights = []
current_cumulative = 0
for model_hash, weight in weights.items():
current_cumulative += weight
cumulative_weights.append((model_hash, current_cumulative))
bucket = hash(str(request_id)) % 100
for model_hash, cumulative in cumulative_weights:
if bucket < cumulative:
return model_hash
logging.warning(f"Routing failed for request {request_id}. No model selected. Weights: {weights}")
return None
class InferenceService:
def __init__(self, model_router, model_base_dir=MODEL_BASE_DIR, model_name=MODEL_NAME_SLUG):
self.model_router = model_router
self.model_base_dir = model_base_dir
self.model_name = model_name
self.loaded_models = {}
def _load_model(self, model_hash):
"""Loads a specific model version from its content-addressable path."""
# The symlink points to the model's *directory*, not the file itself.
# We need to reconstruct the full path.
model_file_path = Path(f"{self.model_base_dir}/{model_hash}/{self.model_name}.joblib")
if not model_file_path.exists():
raise FileNotFoundError(f"Model file not found at expected path: {model_file_path}")
if model_hash not in self.loaded_models:
logging.info(f"Loading model: {model_file_path}")
self.loaded_models[model_hash] = joblib.load(model_file_path)
return self.loaded_models[model_hash]
def predict(self, request_id, input_data):
"""Routes request and performs prediction with health checks."""
model_hash = self.model_router.get_active_model_hash(self.model_name, request_id)
if not model_hash:
logging.error(f"No active model for {self.model_name} to route request {request_id}. Request {request_id} failed.")
return {"request_id": request_id, "error": "No active model available."}
try:
model_pipeline = self._load_model(model_hash)
# --- SIMULATED MODEL DEGRADATION ---
# For model version 2 (v2), simulate a 30% chance of failure
# This is key for demonstrating the automated rollback
if "v2" in model_hash and hash(str(request_id)) % 10 < 3:
raise ValueError("Simulated model degradation: NaN output detected!")
prediction = model_pipeline.predict([input_data]) # Expects a list of strings
return {"request_id": request_id, "model_hash": model_hash[:8], "prediction": prediction.tolist()}
except Exception as e:
logging.error(f"Model {model_hash[:8]} failed for request {request_id}: {e}")
self._handle_model_failure(model_hash)
return {"request_id": request_id, "error": f"Prediction failed for model {model_hash[:8]}: {e}"}
def _handle_model_failure(self, failed_model_hash):
"""Reduces traffic to a failing model and attempts to fall back."""
current_weights = self.model_router.local_cache.get(self.model_name, {})
new_weights = current_weights.copy()
if failed_model_hash in new_weights and new_weights[failed_model_hash] > 0:
logging.warning(f"Reducing traffic to failing model {failed_model_hash[:8]}.")
# Decrease weight of the failing model
reduced_amount = min(new_weights[failed_model_hash], 10) # Reduce by up to 10%
new_weights[failed_model_hash] -= reduced_amount
# Distribute the reduced weight to other active models
other_active_models = [h for h, w in new_weights.items() if h != failed_model_hash and w > 0]
if other_active_models:
share_per_model = reduced_amount / len(other_active_models)
for h in other_active_models:
new_weights[h] += share_per_model
else: # If no other active models, give it back to the first available (e.g., v1)
# This assumes v1 is always the stable fallback
v1_hash = next((h for h in new_weights if "v1" in h), None)
if v1_hash:
new_weights[v1_hash] += reduced_amount
# Normalize weights to sum to 100 (or close) for cleaner percentages
total_weight = sum(new_weights.values())
if total_weight > 0:
new_weights = {k: int(v / total_weight * 100) for k, v in new_weights.items()}
self.model_router.set_weights(self.model_name, new_weights)
else:
logging.info(f"Model {failed_model_hash[:8]} already has 0 traffic or is not in weights. No action needed.")
# --- Main Execution Logic ---
def fetch_and_process_data():
"""Fetches data from RSS feed and prepares it for training."""
try:
feed = feedparser.parse(RSS_FEED_URL)
if feed.bozo: # Check if feed parsing had issues
logging.warning(f"Feed parsing error: {feed.bozo_exception}")
entries = feed.entries
if not entries:
logging.error("No entries found in the RSS feed. Cannot train model.")
return [], []
titles = [entry.title for entry in entries if hasattr(entry, 'title')]
# Simple heuristic to categorize articles for demo purposes
# In a real scenario, you'd have labeled data.
categories = []
for title in titles:
title_lower = title.lower()
if "data" in title_lower or "ml" in title_lower or "ai" in title_lower or "machine learning" in title_lower:
categories.append("ML/Data")
elif "platform" in title_lower or "engineering" in title_lower or "techblog" in title_lower:
categories.append("Platform/Infra")
else:
categories.append("General")
logging.info(f"Fetched {len(titles)} articles from Netflix Tech Blog.")
return titles, categories
except Exception as e:
logging.error(f"Failed to fetch or process RSS feed: {e}")
return [], []
def train_model(X, y, version_tag):
"""Trains a simple text classification pipeline."""
if not X or not y:
logging.error("Cannot train model with empty data.")
return None
pipeline = Pipeline([
('tfidf', TfidfVectorizer(stop_words='english', max_features=1000)),
('clf', LogisticRegression(random_state=42, max_iter=200))
])
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
pipeline.fit(X_train, y_train)
accuracy = accuracy_score(y_test, pipeline.predict(X_test))
logging.info(f"Model {version_tag} trained with accuracy: {accuracy:.2f}")
return pipeline
def cleanup_artifacts():
"""Removes created model directories and Redis keys."""
logging.info(f"Cleaning up {MODEL_BASE_DIR} directory...")
if Path(MODEL_BASE_DIR).exists():
shutil.rmtree(MODEL_BASE_DIR)
try:
r_conn = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True, socket_connect_timeout=1)
r_conn.ping()
logging.info(f"Cleaning up Redis key {MODEL_NAME_SLUG} under {r_conn.hgetall('model_routes')}...")
r_conn.hdel("model_routes", MODEL_NAME_SLUG)
except redis.exceptions.ConnectionError:
logging.warning("Could not connect to Redis for cleanup. Skipping Redis key deletion.")
except Exception as e:
logging.error(f"Error during Redis cleanup: {e}")
def main():
cleanup_artifacts() # Start fresh
# 1. Fetch and prepare data
titles, categories = fetch_and_process_data()
if not titles or not categories:
logging.error("Exiting due to data fetching/processing failure.")
return
# 2. Train and package Model V1 (Production)
logging.info("\n--- Training and packaging Model V1 (Production) ---")
model_v1_pipeline = train_model(titles, categories, "v1")
if not model_v1_pipeline: return
model_v1_hash, model_v1_path = package_model(model_v1_pipeline, "v1")
# 3. Activate Model V1 as initial production
activate_model(model_v1_hash)
# 4. Initialize Model Router and Inference Service
router = ModelRouter()
inference_service = InferenceService(router)
# 5. Set initial weights: 100% to V1
router.set_weights(MODEL_NAME_SLUG, {model_v1_hash: 100})
time.sleep(1) # Give Redis a moment to sync
logging.info("\n--- Simulating initial traffic (100% to V1) ---")
traffic_counts = {model_v1_hash: 0, "errors": 0}
for i in range(NUM_SIMULATED_REQUESTS // 2):
request_id = f"req_init_{i}"
result = inference_service.predict(request_id, titles[i % len(titles)])
if "error" in result:
traffic_counts["errors"] += 1
else:
traffic_counts[result["model_hash"]] = traffic_counts.get(result["model_hash"], 0) + 1
logging.info(f"Initial Traffic Report (V1 only): {traffic_counts}")
# 6. Train and package Model V2 (Canary)
logging.info("\n--- Training and packaging Model V2 (Canary) ---")
# Simulate a slightly different model for v2
model_v2_pipeline = train_model(titles, categories, "v2")
if not model_v2_pipeline: return
model_v2_hash, model_v2_path = package_model(model_v2_pipeline, "v2")
# 7. Introduce V2 with canary traffic (e.g., 90% V1, 10% V2)
logging.info("\n--- Introducing Model V2 with 10% canary traffic ---")
router.set_weights(MODEL_NAME_SLUG, {model_v1_hash: 90, model_v2_hash: 10})
time.sleep(1)
logging.info("\n--- Simulating canary traffic (90% V1, 10% V2 with V2 errors) ---")
canary_traffic_counts = {model_v1_hash: 0, model_v2_hash: 0, "errors": 0}
for i in range(NUM_SIMULATED_REQUESTS):
request_id = f"req_canary_{i}"
result = inference_service.predict(request_id, titles[i % len(titles)])
if "error" in result:
canary_traffic_counts["errors"] += 1
else:
canary_traffic_counts[result["model_hash"]] = canary_traffic_counts.get(result["model_hash"], 0) + 1
# Check current weights after each request to observe dynamic changes
if i % (NUM_SIMULATED_REQUESTS // 10) == 0:
current_weights = router.local_cache.get(MODEL_NAME_SLUG, {})
logging.info(f"Request {i}: Current weights: {current_weights}")
time.sleep(0.1) # Briefly pause to let logs catch up
logging.info(f"\nCanary Traffic Report (after V2 errors and potential rollback):")
final_weights = router.local_cache.get(MODEL_NAME_SLUG, {})
logging.info(f"Final Routing Weights: {final_weights}")
logging.info(f"Traffic distribution: {canary_traffic_counts}")
if final_weights.get(model_v2_hash, 0) < 10: # If traffic reduced
logging.info("SUCCESS: Automated rollback/traffic reduction for Model V2 detected due to simulated errors!")
else:
logging.warning("WARNING: Model V2 traffic was not significantly reduced despite simulated errors.")
logging.info("\n--- Demonstration Complete ---")
cleanup_artifacts()
if __name__ == "__main__":
main()
Expected Output
When you run the script, you should see output similar to this (exact hashes and categories will vary, but the pattern of traffic distribution and rollback should be consistent):INFO - Cleaning up models_deployment directory...
INFO - Cleaning up Redis key tech_classifier under {}...
INFO - Successfully connected to Redis.
INFO - --- Training and packaging Model V1 (Production) ---
INFO - Fetched 50 articles from Netflix Tech Blog.
INFO - Model v1 trained with accuracy: 0.70
INFO - Packaged model v1 to models_deployment/a1b2c3d4e5... (hash: a1b2c3d4)
INFO - Activated model a1b2c3d4 via symlink models_deployment/current_classifier
INFO - Set weights for tech_classifier: {'a1b2c3d4e5...': 100}
INFO - --- Simulating initial traffic (100% to V1) ---
INFO - Model a1b2c3d4 failed for request req_init_2: 'list' object has no attribute 'lower' # Example of a data processing error
ERROR - Model a1b2c3d4 failed for request req_init_2: 'list' object has no attribute 'lower'
WARNING - Reducing traffic to failing model a1b2c3d4.
INFO - Set weights for tech_classifier: {'a1b2c3d4e5...': 90}
INFO - Initial Traffic Report (V1 only): {'a1b2c3d4': 48, 'errors': 2}
INFO - --- Training and packaging Model V2 (Canary) ---
INFO - Model v2 trained with accuracy: 0.72
INFO - Packaged model v2 to models_deployment/f6g7h8i9j0... (hash: f6g7h8i9)
INFO - --- Introducing Model V2 with 10% canary traffic ---
INFO - Set weights for tech_classifier: {'a1b2c3d4e5...': 90, 'f6g7h8i9j0...': 10}
INFO - --- Simulating canary traffic (90% V1, 10% V2 with V2 errors) ---
INFO - Request 0: Current weights: {'a1b2c3d4e5...': 90, 'f6g7h8i9j0...': 10}
ERROR - Model f6g7h8i9 failed for request req_canary_2: Simulated model degradation: NaN output detected!
WARNING - Reducing traffic to failing model f6g7h8i9.
INFO - Set weights for tech_classifier: {'a1b2c3d4e5...': 99, 'f6g7h8i9j0...': 1}
INFO - Request 10: Current weights: {'a1b2c3d4e5...': 99, 'f6g7h8i9j0...': 1}
ERROR - Model f6g7h8i9 failed for request req_canary_12: Simulated model degradation: NaN output detected!
WARNING - Reducing traffic to failing model f6g7h8i9.
INFO - Set weights for tech_classifier: {'a1b2c3d4e5...': 100, 'f6g7h8i9j0...': 0}
INFO - Request 20: Current weights: {'a1b2c3d4e5...': 100, 'f6g7h8i9j0...': 0}
... (more requests, V2 traffic will likely remain at 0) ...
Canary Traffic Report (after V2 errors and potential rollback):
INFO - Final Routing Weights: {'a1b2c3d4e5...': 100, 'f6g7h8i9j0...': 0}
INFO - Traffic distribution: {'a1b2c3d4': 140, 'f6g7h8i9': 2, 'errors': 8}
INFO - SUCCESS: Automated rollback/traffic reduction for Model V2 detected due to simulated errors!
INFO - --- Demonstration Complete ---
INFO - Cleaning up models_deployment directory...
INFO - Cleaning up Redis key tech_classifier under {'model_routes': '{"tech_classifier": "{\\"a1b2c3d4e5...\\": 100, \\"f6g7h8i9j0...\\": 0}"}'}...