CloudNativePG : Gestion automatisée et résiliente des clusters PostgreSQL sur Kubernetes : haute disponibilité, sauvegarde, réplication, intégration TLS, et DevOps déclaratif multi-environnements
CloudNativePG est un opérateur open source conçu pour orchestrer des charges PostgreSQL sur des clusters Kubernetes. Il repose sur les principes DevOps, notamment la configuration déclarative et…
coolibra08 juil. 20261125 vues
CloudNativePG est un opérateur open source conçu pour orchestrer des charges PostgreSQL sur des clusters Kubernetes. Il repose sur les principes DevOps, notamment la configuration déclarative et l’infrastructure immuable.
L’opérateur introduit la ressource Kubernetes Cluster, représentant un cluster PostgreSQL constitué d’un nœud principal et de répliques facultatives, offrant haute disponibilité et déchargement des requêtes en lecture seule. Les applications internes au cluster Kubernetes accèdent à PostgreSQL via un service géré par l’opérateur, garantissant transparence lors des basculements ou changements de rôle principal. Pour les applications externes, un objet Service ou Ingress est nécessaire.
CloudNativePG, développé initialement par EDB, est publié sous licence Apache 2.0 et a été proposé à la CNCF Sandbox en avril 2022. Github repo
Caractéristiques principales :
- Haute disponibilité et auto-guérison : promotion automatique de répliques en cas de défaillance.
- Gestion déclarative : déploiement et configuration via des spécifications Kubernetes.
- Mises à jour simplifiées : versions mineures ou majeures de PostgreSQL et de l’opérateur.
- Sauvegarde et restauration : intégration avec AWS S3, Azure Blob Storage, et Google Cloud Storage, récupération complète ou à un moment précis.
- Sécurité renforcée : connexions TLS, certificats personnalisés, et authentification LDAP.
- Extensibilité : support de PostGIS, affinités de nœuds, et intégration PgBouncer pour le pooling de connexions.
Requirements:
- Cluster kubernetes fonctionel
- Installer helm sur le post admin
- Le cluster accessible depuis le laptop admin
Installer l'operateur CloudNativePG
- Ajout du chart helm cloudnative-pg
helm repo add cnpg https://cloudnative-pg.github.io/charts
helm repo update
- Installer l'opérateur CloudNativePG
- Par defaut, l'installation est multi namespaces. Pour signle namespace, il faut mettre ce parametre à false config.clusterWide=false
helm upgrade --install cnpg \
--namespace cnpg-system \
--create-namespace \
cnpg/cloudnative-pg
- Vérifier l'installation
kubectl get po -n cnpg-system
Déployer un cluster postgres
- Cluster postgres avec le crd
- Exemple complet de cluster crd
apiVersion: v1
kind: Namespace
metadata:
name: cluster-postgres-dev
---
apiVersion: postgresql.cnpg.io/v1
kind: Cluster
metadata:
name: cluster-dev
namespace: cluster-postgres-dev
spec:
instances: 3
storage:
size: 1Gi
-
- Appliquer le manifest yaml
kubectl apply -f cluster-crd.yaml
- Regroupement de connexions: PgBouncer
PgBouncer est un outil proxy qui gère les connexions entre les applications clientes et un cluster PostgreSQL. Il agit comme un intermédiaire pour réduire la surcharge liée à la gestion des connexions par PostgreSQL.
Problème résolu par PgBouncer :
PostgreSQL gère chaque connexion avec un processus dédié (modèle "process per connection"), ce qui peut consommer beaucoup de ressources lorsque :
- Il y a de nombreuses connexions actives simultanées.
- Les connexions sont courtes et fréquentes (par exemple, dans les environnements web ou microservices).
Impact : Chaque connexion nécessite une allocation mémoire et des ressources, ce qui peut conduire à une saturation rapide du serveur.
- Cette config permet de déployer pgbouncer
spec:
type: rw
pgbouncer:
poolMode: session
parameters:
max_client_conn: "1000"
default_pool_size: "10"
-
- Vérification
kubectl get secret,svc,po -n cluster-postgres-dev -o wide
Le secret cluster-dev-app contient les paramètres de connexion au cluster postgres
-
- Recuperer les parametres d'accès dans le secret: cluster-dev-app
kubectl -n cluster-postgres-dev edit secrets cluster-dev-app
# pour decoder le base64
echo "cG9zdGdyZXNxbDovL2FwcDozbXhvcVozb2JPb" | base64 -d
- Cluster postgres avec le chart helm
-
- Ajout du chart helm cloudnative-pg
helm repo add cnpg https://cloudnative-pg.github.io/charts
helm repo update
-
- Déployer un cluster postgres
helm upgrade --install cluster-postgres-dev \
--namespace cluster-postgres-dev \
--create-namespace \
cnpg/cluster
- Cluster avec pooler et limit resources
apiVersion: postgresql.cnpg.io/v1
kind: Cluster
metadata:
name: cluster-dev
namespace: cluster-postgres-dev
spec:
instances: 3
storage:
storageClass: local-path-rwx
size: 1Gi
# managed:
# services:
# ## disable the default services
# disabledDefaultServices: ["ro", "r"]
resources:
requests:
memory: "512Mi"
cpu: "200m"
limits:
memory: "1Gi"
cpu: "1000m"
---
apiVersion: postgresql.cnpg.io/v1
kind: Pooler
metadata:
name: cluster-dev-pooler-rw
namespace: cluster-postgres-dev
spec:
cluster:
name: cluster-dev
instances: 3
type: rw
serviceTemplate:
metadata:
name: "svc-lb-pooler-rw"
spec:
type: LoadBalancer
loadBalancerIP: 10.10.10.20
pgbouncer:
poolMode: session
parameters:
max_client_conn: "1000"
default_pool_size: "10"
- Test de performances avec python
#!/usr/bin/env python3
"""
Script de test de performances pour un cluster CloudNative PostgreSQL
Utilise SQLAlchemy comme ORM et génère des graphiques de performances
"""
import argparse
import time
import datetime
import random
import string
import concurrent.futures
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Float, ForeignKey, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.pool import QueuePool
# Configuration par défaut
DEFAULT_HOST = "10.10.10.20" # IP du LoadBalancer du pooler
DEFAULT_PORT = "5432"
DEFAULT_DB = "postgres"
DEFAULT_USER = "postgres"
DEFAULT_PASSWORD = "postgres"
DEFAULT_CONCURRENCY = 10
DEFAULT_DURATION = 60 # durée du test en secondes
# Création du modèle ORM
Base = declarative_base()
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key=True)
name = Column(String(100))
email = Column(String(100))
registration_date = Column(DateTime, default=datetime.datetime.utcnow)
orders = relationship("Order", back_populates="customer")
def __repr__(self):
return f"<Customer(id={self.id}, name='{self.name}')>"
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String(100))
description = Column(String(500))
price = Column(Float)
order_items = relationship("OrderItem", back_populates="product")
def __repr__(self):
return f"<Product(id={self.id}, name='{self.name}', price={self.price})>"
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
customer_id = Column(Integer, ForeignKey('customers.id'))
order_date = Column(DateTime, default=datetime.datetime.utcnow)
total_amount = Column(Float)
customer = relationship("Customer", back_populates="orders")
items = relationship("OrderItem", back_populates="order")
def __repr__(self):
return f"<Order(id={self.id}, customer_id={self.customer_id}, total={self.total_amount})>"
class OrderItem(Base):
__tablename__ = 'order_items'
id = Column(Integer, primary_key=True)
order_id = Column(Integer, ForeignKey('orders.id'))
product_id = Column(Integer, ForeignKey('products.id'))
quantity = Column(Integer)
price = Column(Float)
order = relationship("Order", back_populates="items")
product = relationship("Product", back_populates="order_items")
def __repr__(self):
return f"<OrderItem(order_id={self.order_id}, product_id={self.product_id}, quantity={self.quantity})>"
# Fonctions pour générer des données aléatoires
def random_string(length=10):
"""Génère une chaîne aléatoire de la longueur spécifiée"""
letters = string.ascii_lowercase
return ''.join(random.choice(letters) for _ in range(length))
def random_email():
"""Génère un email aléatoire"""
name = random_string(8)
domain = random_string(6)
return f"{name}@{domain}.com"
def random_price():
"""Génère un prix aléatoire entre 10 et 1000"""
return round(random.uniform(10, 1000), 2)
# Fonctions pour les opérations de test
def create_schema(engine):
"""Crée le schéma de la base de données"""
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
def insert_data(session, num_customers=100, num_products=200):
"""Insère des données initiales dans la base"""
# Insertion des clients
customers = []
for _ in range(num_customers):
customer = Customer(
name=f"{random_string(6)} {random_string(8)}",
email=random_email()
)
customers.append(customer)
session.add_all(customers)
# Insertion des produits
products = []
for _ in range(num_products):
product = Product(
name=random_string(10),
description=random_string(50),
price=random_price()
)
products.append(product)
session.add_all(products)
session.commit()
return customers, products
def perform_read_test(session):
"""Exécute un test de lecture aléatoire"""
start_time = time.time()
# Sélectionne aléatoirement un type de requête
query_type = random.choice(['customer', 'product', 'order', 'join'])
if query_type == 'customer':
# Obtenir un client aléatoire
customer_id = random.randint(1, 100)
result = session.query(Customer).filter(Customer.id == customer_id).first()
elif query_type == 'product':
# Obtenir un produit aléatoire
product_id = random.randint(1, 200)
result = session.query(Product).filter(Product.id == product_id).first()
elif query_type == 'order':
# Obtenir une commande aléatoire
if random.random() < 0.5 and session.query(Order).count() > 0:
order_id = session.query(Order.id).order_by(text('random()')).first()[0]
result = session.query(Order).filter(Order.id == order_id).first()
else:
result = None
else:
# Exécuter une jointure
if session.query(Order).count() > 0:
result = session.query(Customer, Order).join(Order).order_by(text('random()')).limit(10).all()
else:
result = []
end_time = time.time()
return end_time - start_time
def perform_write_test(session, customers, products):
"""Exécute un test d'écriture aléatoire"""
start_time = time.time()
# Sélectionne aléatoirement un type d'opération d'écriture
write_type = random.choice(['new_order', 'update_product', 'new_customer'])
if write_type == 'new_order':
# Créer une nouvelle commande
customer = random.choice(customers)
num_items = random.randint(1, 5)
total_amount = 0
order = Order(customer_id=customer.id, total_amount=0)
session.add(order)
session.flush() # Pour obtenir l'ID de la commande
for _ in range(num_items):
product = random.choice(products)
quantity = random.randint(1, 5)
price = product.price
item_total = price * quantity
total_amount += item_total
order_item = OrderItem(
order_id=order.id,
product_id=product.id,
quantity=quantity,
price=price
)
session.add(order_item)
order.total_amount = total_amount
elif write_type == 'update_product':
# Mettre à jour un produit
product = random.choice(products)
product.price = random_price()
product.description = random_string(50)
else:
# Créer un nouveau client
customer = Customer(
name=f"{random_string(6)} {random_string(8)}",
email=random_email()
)
session.add(customer)
customers.append(customer)
session.commit()
end_time = time.time()
return end_time - start_time
def worker(args, worker_id):
"""Fonction exécutée par chaque thread/processus"""
# Configuration de la connexion à la base de données
conn_str = f"postgresql://{args.user}:{args.password}@{args.host}:{args.port}/{args.dbname}"
engine = create_engine(
conn_str,
poolclass=QueuePool,
pool_size=5,
max_overflow=10
)
Session = sessionmaker(bind=engine)
session = Session()
# Résultats pour ce worker
results = {
'read_times': [],
'write_times': [],
'errors': 0
}
# Si c'est le premier worker, initialiser la base de données
if worker_id == 0:
try:
print(f"Worker {worker_id}: Initialisation du schéma et des données...")
create_schema(engine)
customers, products = insert_data(session)
except Exception as e:
print(f"Erreur lors de l'initialisation: {e}")
return results
else:
# Attendre un peu pour que le premier worker initialise la base
time.sleep(2)
customers = session.query(Customer).all()
products = session.query(Product).all()
start_time = time.time()
end_time = start_time + args.duration
# Ratio lecture/écriture (80% lectures, 20% écritures)
read_write_ratio = 0.8
operation_count = 0
try:
while time.time() < end_time:
try:
# Décider si c'est une opération de lecture ou d'écriture
if random.random() < read_write_ratio:
# Opération de lecture
read_time = perform_read_test(session)
results['read_times'].append(read_time)
else:
# Opération d'écriture
write_time = perform_write_test(session, customers, products)
results['write_times'].append(write_time)
operation_count += 1
# Petit délai pour éviter de surcharger
if operation_count % 10 == 0:
time.sleep(0.01)
except Exception as e:
results['errors'] += 1
print(f"Worker {worker_id} a rencontré une erreur: {e}")
# Recréer une session en cas d'erreur
session.close()
session = Session()
finally:
session.close()
return results
def generate_graphs(all_results):
"""Génère les graphiques de performances"""
# Préparer les données
all_read_times = [time for worker_result in all_results for time in worker_result['read_times']]
all_write_times = [time for worker_result in all_results for time in worker_result['write_times']]
total_errors = sum(worker_result['errors'] for worker_result in all_results)
# Calculer des statistiques
stats = {
'read': {
'count': len(all_read_times),
'min': min(all_read_times) if all_read_times else 0,
'max': max(all_read_times) if all_read_times else 0,
'avg': sum(all_read_times) / len(all_read_times) if all_read_times else 0,
'p95': np.percentile(all_read_times, 95) if all_read_times else 0,
'p99': np.percentile(all_read_times, 99) if all_read_times else 0
},
'write': {
'count': len(all_write_times),
'min': min(all_write_times) if all_write_times else 0,
'max': max(all_write_times) if all_write_times else 0,
'avg': sum(all_write_times) / len(all_write_times) if all_write_times else 0,
'p95': np.percentile(all_write_times, 95) if all_write_times else 0,
'p99': np.percentile(all_write_times, 99) if all_write_times else 0
},
'errors': total_errors
}
# Créer une figure avec plusieurs sous-graphiques
plt.figure(figsize=(15, 10))
# 1. Histogramme des temps de réponse pour les lectures
plt.subplot(2, 2, 1)
plt.hist(all_read_times, bins=50, alpha=0.7, color='blue')
plt.title('Distribution des temps de réponse (lectures)')
plt.xlabel('Temps (s)')
plt.ylabel('Nombre de requêtes')
plt.grid(True, linestyle='--', alpha=0.7)
# 2. Histogramme des temps de réponse pour les écritures
plt.subplot(2, 2, 2)
plt.hist(all_write_times, bins=50, alpha=0.7, color='green')
plt.title('Distribution des temps de réponse (écritures)')
plt.xlabel('Temps (s)')
plt.ylabel('Nombre de requêtes')
plt.grid(True, linestyle='--', alpha=0.7)
# 3. Graphique à barres comparant les performances de lecture et d'écriture
plt.subplot(2, 2, 3)
metrics = ['min', 'avg', 'p95', 'p99', 'max']
read_values = [stats['read'][m] for m in metrics]
write_values = [stats['write'][m] for m in metrics]
x = np.arange(len(metrics))
width = 0.35
plt.bar(x - width/2, read_values, width, label='Lecture', color='blue')
plt.bar(x + width/2, write_values, width, label='Écriture', color='green')
plt.xlabel('Métrique')
plt.ylabel('Temps (s)')
plt.title('Comparaison des performances lecture/écriture')
plt.xticks(x, metrics)
plt.legend()
plt.grid(True, linestyle='--', alpha=0.7)
# 4. Tableau résumé des statistiques
plt.subplot(2, 2, 4)
plt.axis('off')
table_data = [
['Métrique', 'Lecture', 'Écriture'],
['Nombre de requêtes', stats['read']['count'], stats['write']['count']],
['Temps min (ms)', f"{stats['read']['min']*1000:.2f}", f"{stats['write']['min']*1000:.2f}"],
['Temps moyen (ms)', f"{stats['read']['avg']*1000:.2f}", f"{stats['write']['avg']*1000:.2f}"],
['p95 (ms)', f"{stats['read']['p95']*1000:.2f}", f"{stats['write']['p95']*1000:.2f}"],
['p99 (ms)', f"{stats['read']['p99']*1000:.2f}", f"{stats['write']['p99']*1000:.2f}"],
['Temps max (ms)', f"{stats['read']['max']*1000:.2f}", f"{stats['write']['max']*1000:.2f}"],
['Erreurs totales', stats['errors'], '']
]
plt.table(cellText=table_data, loc='center', cellLoc='center', colWidths=[0.3, 0.3, 0.3])
plt.title('Résumé des performances')
plt.tight_layout()
plt.savefig('performance_results.png', dpi=300)
# Créer un DataFrame et le sauvegarder
results_df = pd.DataFrame({
'Type': ['Lecture'] * len(all_read_times) + ['Écriture'] * len(all_write_times),
'Temps (s)': all_read_times + all_write_times
})
results_df.to_csv('performance_results.csv', index=False)
print("\nRésultats de performance sauvegardés dans 'performance_results.png' et 'performance_results.csv'")
print("\nStatistiques sommaires:")
print(f"Nombre total de requêtes: {stats['read']['count'] + stats['write']['count']}")
print(f"Erreurs: {stats['errors']}")
print("\nTempss de lecture (s):")
print(f" Min: {stats['read']['min']:.6f}")
print(f" Avg: {stats['read']['avg']:.6f}")
print(f" P95: {stats['read']['p95']:.6f}")
print(f" P99: {stats['read']['p99']:.6f}")
print(f" Max: {stats['read']['max']:.6f}")
print("\nTemps d'écriture (s):")
print(f" Min: {stats['write']['min']:.6f}")
print(f" Avg: {stats['write']['avg']:.6f}")
print(f" P95: {stats['write']['p95']:.6f}")
print(f" P99: {stats['write']['p99']:.6f}")
print(f" Max: {stats['write']['max']:.6f}")
def main():
parser = argparse.ArgumentParser(description='Test de performances pour un cluster CloudNative PostgreSQL')
parser.add_argument('--host', default=DEFAULT_HOST, help=f'Hôte PostgreSQL (défaut: {DEFAULT_HOST})')
parser.add_argument('--port', default=DEFAULT_PORT, help=f'Port PostgreSQL (défaut: {DEFAULT_PORT})')
parser.add_argument('--dbname', default=DEFAULT_DB, help=f'Nom de la base de données (défaut: {DEFAULT_DB})')
parser.add_argument('--user', default=DEFAULT_USER, help=f'Utilisateur PostgreSQL (défaut: {DEFAULT_USER})')
parser.add_argument('--password', default=DEFAULT_PASSWORD, help=f'Mot de passe PostgreSQL (défaut: {DEFAULT_PASSWORD})')
parser.add_argument('--concurrency', type=int, default=DEFAULT_CONCURRENCY, help=f'Nombre de connexions concurrentes (défaut: {DEFAULT_CONCURRENCY})')
parser.add_argument('--duration', type=int, default=DEFAULT_DURATION, help=f'Durée du test en secondes (défaut: {DEFAULT_DURATION})')
args = parser.parse_args()
print(f"Démarrage du test de performances PostgreSQL sur {args.host}:{args.port}/{args.dbname}")
print(f"Concurrence: {args.concurrency} | Durée: {args.duration} secondes")
all_results = []
# Utiliser un pool de threads pour la concurrence
with concurrent.futures.ThreadPoolExecutor(max_workers=args.concurrency) as executor:
# Soumettre les tâches
future_to_worker = {executor.submit(worker, args, i): i for i in range(args.concurrency)}
# Collecter les résultats
for future in concurrent.futures.as_completed(future_to_worker):
worker_id = future_to_worker[future]
try:
result = future.result()
all_results.append(result)
print(f"Worker {worker_id} terminé: {len(result['read_times'])} lectures, {len(result['write_times'])} écritures, {result['errors']} erreurs")
except Exception as e:
print(f"Worker {worker_id} a échoué: {e}")
# Générer les graphiques
generate_graphs(all_results)
if __name__ == "__main__":
main()
- Builde le Dockerfile
- requirements.txt
sqlalchemy>=1.4.0
matplotlib>=3.5.0
pandas>=1.3.0
numpy>=1.20.0
psycopg2-binary>=2.9.0
-
- Dockerfile
FROM python:3.9.22-slim
RUN apt-get update && apt-get install -y \
gcc \
python3-dev \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY benchmark.py .
RUN mkdir -p /app/results
# Définition du volume pour conserver les résultats
VOLUME /app/results
# Port pour une éventuelle interface web future
EXPOSE 8050
-
- Build de l'image et lancer les tests
docker build -t pg-perf-tester .
docker run --rm -it -v ./perf/results:/app/results pg-perf:0.0.1 bash
python ../benchmark.py --host=10.10.10.20 --port=5432 --dbname=app --user=app \
--password=YhJ4bHgX21kGqGwMWqMXVJbYo4GFz6BOgQO0pjULuVTaMNJjG8n75AKr2nnUsbBP \
--concurrency=10 --duration=60
Commentaires
Connectez-vous pour rejoindre la discussion.
Aucun commentaire pour le moment.