Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

29 NODES
๐Ÿค–system promptโ€ข7 months ago

api-design-principles

Master REST and GraphQL API design principles to build intuitive,

coding
โญ1
# API Design Principles Master REST and GraphQL API design principles to build intuitive, scalable, and maintainable APIs that delight developers and stand the test of time. ## When to Use This Skill - Designing new REST or GraphQL APIs - Refactoring existing APIs for better usability - Establishing API design standards for your team - Reviewing API specifications before implementation - Migrating between API paradigms (REST to GraphQL, etc.) - Creating developer-friendly API documentation - Optimizing APIs for specific use cases (mobile, third-party integrations) ## Core Concepts ### 1. RESTful Design Principles **Resource-Oriented Architecture** - Resources are nouns (users, orders, products), not verbs - Use HTTP methods for actions (GET, POST, PUT, PATCH, DELETE) - URLs represent resource hierarchies - Consistent naming conventions **HTTP Methods Semantics:** - `GET`: Retrieve resources (idempotent, safe) - `POST`: Create new resources - `PUT`: Replace entire resource (idempotent) - `PATCH`: Partial resource updates - `DELETE`: Remove resources (idempotent) ### 2. GraphQL Design Principles **Schema-First Development** - Types define your domain model - Queries for reading data - Mutations for modifying data - Subscriptions for real-time updates **Query Structure:** - Clients request exactly what they need - Single endpoint, multiple operations - Strongly typed schema - Introspection built-in ### 3. API Versioning Strategies **URL Versioning:** ``` /api/v1/users /api/v2/users ``` **Header Versioning:** ``` Accept: application/vnd.api+json; version=1 ``` **Query Parameter Versioning:** ``` /api/users?version=1 ``` ## REST API Design Patterns ### Pattern 1: Resource Collection Design ```python # Good: Resource-oriented endpoints GET /api/users # List users (with pagination) POST /api/users # Create user GET /api/users/{id} # Get specific user PUT /api/users/{id} # Replace user PATCH /api/users/{id} # Update user fields DELETE /api/users/{id} # Delete user # Nested resources GET /api/users/{id}/orders # Get user's orders POST /api/users/{id}/orders # Create order for user # Bad: Action-oriented endpoints (avoid) POST /api/createUser POST /api/getUserById POST /api/deleteUser ``` ### Pattern 2: Pagination and Filtering ```python from typing import List, Optional from pydantic import BaseModel, Field class PaginationParams(BaseModel): page: int = Field(1, ge=1, description="Page number") page_size: int = Field(20, ge=1, le=100, description="Items per page") class FilterParams(BaseModel): status: Optional[str] = None created_after: Optional[str] = None search: Optional[str] = None class PaginatedResponse(BaseModel): items: List[dict] total: int page: int page_size: int pages: int @property def has_next(self) -> bool: return self.page < self.pages @property def has_prev(self) -> bool: return self.page > 1 # FastAPI endpoint example from fastapi import FastAPI, Query, Depends app = FastAPI() @app.get("/api/users", response_model=PaginatedResponse) async def list_users( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), search: Optional[str] = Query(None) ): # Apply filters query = build_query(status=status, search=search) # Count total total = await count_users(query) # Fetch page offset = (page - 1) * page_size users = await fetch_users(query, limit=page_size, offset=offset) return PaginatedResponse( items=users, total=total, page=page, page_size=page_size, pages=(total + page_size - 1) // page_size ) ``` ### Pattern 3: Error Handling and Status Codes ```python from fastapi import HTTPException, status from pydantic import BaseModel class ErrorResponse(BaseModel): error: str message: str details: Optional[dict] = None timestamp: str path: str class ValidationErrorDetail(BaseModel): field: str message: str value: Any # Consistent error responses STATUS_CODES = { "success": 200, "created": 201, "no_content": 204, "bad_request": 400, "unauthorized": 401, "forbidden": 403, "not_found": 404, "conflict": 409, "unprocessable": 422, "internal_error": 500 } def raise_not_found(resource: str, id: str): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": "NotFound", "message": f"{resource} not found", "details": {"id": id} } ) def raise_validation_error(errors: List[ValidationErrorDetail]): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "error": "ValidationError", "message": "Request validation failed", "details": {"errors": [e.dict() for e in errors]} } ) # Example usage @app.get("/api/users/{user_id}") async def get_user(user_id: str): user = await fetch_user(user_id) if not user: raise_not_found("User", user_id) return user ``` ### Pattern 4: HATEOAS (Hypermedia as the Engine of Application State) ```python class UserResponse(BaseModel): id: str name: str email: str _links: dict @classmethod def from_user(cls, user: User, base_url: str): return cls( id=user.id, name=user.name, email=user.email, _links={ "self": {"href": f"{base_url}/api/users/{user.id}"}, "orders": {"href": f"{base_url}/api/users/{user.id}/orders"}, "update": { "href": f"{base_url}/api/users/{user.id}", "method": "PATCH" }, "delete": { "href": f"{base_url}/api/users/{user.id}", "method": "DELETE" } } ) ``` ## GraphQL Design Patterns ### Pattern 1: Schema Design ```graphql # schema.graphql # Clear type definitions type User { id: ID! email: String! name: String! createdAt: DateTime! # Relationships orders(first: Int = 20, after: String, status: OrderStatus): OrderConnection! profile: UserProfile } type Order { id: ID! status: OrderStatus! total: Money! items: [OrderItem!]! createdAt: DateTime! # Back-reference user: User! } # Pagination pattern (Relay-style) type OrderConnection { edges: [OrderEdge!]! pageInfo: PageInfo! totalCount: Int! } type OrderEdge { node: Order! cursor: String! } type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String } # Enums for type safety enum OrderStatus { PENDING CONFIRMED SHIPPED DELIVERED CANCELLED } # Custom scalars scalar DateTime scalar Money # Query root type Query { user(id: ID!): User users(first: Int = 20, after: String, search: String): UserConnection! order(id: ID!): Order } # Mutation root type Mutation { createUser(input: CreateUserInput!): CreateUserPayload! updateUser(input: UpdateUserInput!): UpdateUserPayload! deleteUser(id: ID!): DeleteUserPayload! createOrder(input: CreateOrderInput!): CreateOrderPayload! } # Input types for mutations input CreateUserInput { email: String! name: String! password: String! } # Payload types for mutations type CreateUserPayload { user: User errors: [Error!] } type Error { field: String message: String! } ``` ### Pattern 2: Resolver Design ```python from typing import Optional, List from ariadne import QueryType, MutationType, ObjectType from dataclasses import dataclass query = QueryType() mutation = MutationType() user_type = ObjectType("User") @query.field("user") async def resolve_user(obj, info, id: str) -> Optional[dict]: """Resolve single user by ID.""" return await fetch_user_by_id(id) @query.field("users") async def resolve_users( obj, info, first: int = 20, after: Optional[str] = None, search: Optional[str] = None ) -> dict: """Resolve paginated user list.""" # Decode cursor offset = decode_cursor(after) if after else 0 # Fetch users users = await fetch_users( limit=first + 1, # Fetch one extra to check hasNextPage offset=offset, search=search ) # Pagination has_next = len(users) > first if has_next: users = users[:first] edges = [ { "node": user, "cursor": encode_cursor(offset + i) } for i, user in enumerate(users) ] return { "edges": edges, "pageInfo": { "hasNextPage": has_next, "hasPreviousPage": offset > 0, "startCursor": edges[0]["cursor"] if edges else None, "endCursor": edges[-1]["cursor"] if edges else None }, "totalCount": await count_users(search=search) } @user_type.field("orders") async def resolve_user_orders(user: dict, info, first: int = 20) -> dict: """Resolve user's orders (N+1 prevention with DataLoader).""" # Use DataLoader to batch requests loader = info.context["loaders"]["orders_by_user"] orders = await loader.load(user["id"]) return paginate_orders(orders, first) @mutation.field("createUser") async def resolve_create_user(obj, info, input: dict) -> dict: """Create new user.""" try: # Validate input validate_user_input(input) # Create user user = await create_user( email=input["email"], name=input["name"], password=hash_password(input["password"]) ) return { "user": user, "errors": [] } except ValidationError as e: return { "user": None, "errors": [{"field": e.field, "message": e.message}] } ``` ### Pattern 3: DataLoader (N+1 Problem Prevention) ```python from aiodataloader import DataLoader from typing import List, Optional class UserLoader(DataLoader): """Batch load users by ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[Optional[dict]]: """Load multiple users in single query.""" users = await fetch_users_by_ids(user_ids) # Map results back to input order user_map = {user["id"]: user for user in users} return [user_map.get(user_id) for user_id in user_ids] class OrdersByUserLoader(DataLoader): """Batch load orders by user ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[List[dict]]: """Load orders for multiple users in single query.""" orders = await fetch_orders_by_user_ids(user_ids) # Group orders by user_id orders_by_user = {} for order in orders: user_id = order["user_id"] if user_id not in orders_by_user: orders_by_user[user_id] = [] orders_by_user[user_id].append(order) # Return in input order return [orders_by_user.get(user_id, []) for user_id in user_ids] # Context setup def create_context(): return { "loaders": { "user": UserLoader(), "orders_by_user": OrdersByUserLoader() } } ``` ## Best Practices ### REST APIs 1. **Consistent Naming**: Use plural nouns for collections (`/users`, not `/user`) 2. **Stateless**: Each request contains all necessary information 3. **Use HTTP Status Codes Correctly**: 2xx success, 4xx client errors, 5xx server errors 4. **Version Your API**: Plan for breaking changes from day one 5. **Pagination**: Always paginate large collections 6. **Rate Limiting**: Protect your API with rate limits 7. **Documentation**: Use OpenAPI/Swagger for interactive docs ### GraphQL APIs 1. **Schema First**: Design schema before writing resolvers 2. **Avoid N+1**: Use DataLoaders for efficient data fetching 3. **Input Validation**: Validate at schema and resolver levels 4. **Error Handling**: Return structured errors in mutation payloads 5. **Pagination**: Use cursor-based pagination (Relay spec) 6. **Deprecation**: Use `@deprecated` directive for gradual migration 7. **Monitoring**: Track query complexity and execution time ## Common Pitfalls - **Over-fetching/Under-fetching (REST)**: Fixed in GraphQL but requires DataLoaders - **Breaking Changes**: Version APIs or use deprecation strategies - **Inconsistent Error Formats**: Standardize error responses - **Missing Rate Limits**: APIs without limits are vulnerable to abuse - **Poor Documentation**: Undocumented APIs frustrate developers - **Ignoring HTTP Semantics**: POST for idempotent operations breaks expectations - **Tight Coupling**: API structure shouldn't mirror database schema ## Resources - **references/rest-best-practices.md**: Comprehensive REST API design guide - **references/graphql-schema-design.md**: GraphQL schema patterns and anti-patterns - **references/api-versioning-strategies.md**: Versioning approaches and migration paths - **assets/rest-api-template.py**: FastAPI REST API template - **assets/graphql-schema-template.graphql**: Complete GraphQL schema example - **assets/api-design-checklist.md**: Pre-implementation review checklist - **scripts/openapi-generator.py**: Generate OpenAPI specs from code
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

event-store-design

Design and implement event stores for event-sourced systems. Use

coding
โญ1
# Event Store Design Comprehensive guide to designing event stores for event-sourced applications. ## When to Use This Skill - Designing event sourcing infrastructure - Choosing between event store technologies - Implementing custom event stores - Optimizing event storage and retrieval - Setting up event store schemas - Planning for event store scaling ## Core Concepts ### 1. Event Store Architecture ``` โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ Event Store โ”‚ โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ โ”‚ โ”‚ Stream 1 โ”‚ โ”‚ Stream 2 โ”‚ โ”‚ Stream 3 โ”‚ โ”‚ โ”‚ โ”‚ (Aggregate) โ”‚ โ”‚ (Aggregate) โ”‚ โ”‚ (Aggregate) โ”‚ โ”‚ โ”‚ โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”‚ โ”‚ โ”‚ Event 1 โ”‚ โ”‚ Event 1 โ”‚ โ”‚ Event 1 โ”‚ โ”‚ โ”‚ โ”‚ Event 2 โ”‚ โ”‚ Event 2 โ”‚ โ”‚ Event 2 โ”‚ โ”‚ โ”‚ โ”‚ Event 3 โ”‚ โ”‚ ... โ”‚ โ”‚ Event 3 โ”‚ โ”‚ โ”‚ โ”‚ ... โ”‚ โ”‚ โ”‚ โ”‚ Event 4 โ”‚ โ”‚ โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”‚ Global Position: 1 โ†’ 2 โ†’ 3 โ†’ 4 โ†’ 5 โ†’ 6 โ†’ ... โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ ``` ### 2. Event Store Requirements | Requirement | Description | | ----------------- | ---------------------------------- | | **Append-only** | Events are immutable, only appends | | **Ordered** | Per-stream and global ordering | | **Versioned** | Optimistic concurrency control | | **Subscriptions** | Real-time event notifications | | **Idempotent** | Handle duplicate writes safely | ## Technology Comparison | Technology | Best For | Limitations | | ---------------- | ------------------------- | -------------------------------- | | **EventStoreDB** | Pure event sourcing | Single-purpose | | **PostgreSQL** | Existing Postgres stack | Manual implementation | | **Kafka** | High-throughput streaming | Not ideal for per-stream queries | | **DynamoDB** | Serverless, AWS-native | Query limitations | | **Marten** | .NET ecosystems | .NET specific | ## Templates ### Template 1: PostgreSQL Event Store Schema ```sql -- Events table CREATE TABLE events ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), stream_id VARCHAR(255) NOT NULL, stream_type VARCHAR(255) NOT NULL, event_type VARCHAR(255) NOT NULL, event_data JSONB NOT NULL, metadata JSONB DEFAULT '{}', version BIGINT NOT NULL, global_position BIGSERIAL, created_at TIMESTAMPTZ DEFAULT NOW(), CONSTRAINT unique_stream_version UNIQUE (stream_id, version) ); -- Index for stream queries CREATE INDEX idx_events_stream_id ON events(stream_id, version); -- Index for global subscription CREATE INDEX idx_events_global_position ON events(global_position); -- Index for event type queries CREATE INDEX idx_events_event_type ON events(event_type); -- Index for time-based queries CREATE INDEX idx_events_created_at ON events(created_at); -- Snapshots table CREATE TABLE snapshots ( stream_id VARCHAR(255) PRIMARY KEY, stream_type VARCHAR(255) NOT NULL, snapshot_data JSONB NOT NULL, version BIGINT NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW() ); -- Subscriptions checkpoint table CREATE TABLE subscription_checkpoints ( subscription_id VARCHAR(255) PRIMARY KEY, last_position BIGINT NOT NULL DEFAULT 0, updated_at TIMESTAMPTZ DEFAULT NOW() ); ``` ### Template 2: Python Event Store Implementation ```python from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional, List from uuid import UUID, uuid4 import json import asyncpg @dataclass class Event: stream_id: str event_type: str data: dict metadata: dict = field(default_factory=dict) event_id: UUID = field(default_factory=uuid4) version: Optional[int] = None global_position: Optional[int] = None created_at: datetime = field(default_factory=datetime.utcnow) class EventStore: def __init__(self, pool: asyncpg.Pool): self.pool = pool async def append_events( self, stream_id: str, stream_type: str, events: List[Event], expected_version: Optional[int] = None ) -> List[Event]: """Append events to a stream with optimistic concurrency.""" async with self.pool.acquire() as conn: async with conn.transaction(): # Check expected version if expected_version is not None: current = await conn.fetchval( "SELECT MAX(version) FROM events WHERE stream_id = $1", stream_id ) current = current or 0 if current != expected_version: raise ConcurrencyError( f"Expected version {expected_version}, got {current}" ) # Get starting version start_version = await conn.fetchval( "SELECT COALESCE(MAX(version), 0) + 1 FROM events WHERE stream_id = $1", stream_id ) # Insert events saved_events = [] for i, event in enumerate(events): event.version = start_version + i row = await conn.fetchrow( """ INSERT INTO events (id, stream_id, stream_type, event_type, event_data, metadata, version, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING global_position """, event.event_id, stream_id, stream_type, event.event_type, json.dumps(event.data), json.dumps(event.metadata), event.version, event.created_at ) event.global_position = row['global_position'] saved_events.append(event) return saved_events async def read_stream( self, stream_id: str, from_version: int = 0, limit: int = 1000 ) -> List[Event]: """Read events from a stream.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ SELECT id, stream_id, event_type, event_data, metadata, version, global_position, created_at FROM events WHERE stream_id = $1 AND version >= $2 ORDER BY version LIMIT $3 """, stream_id, from_version, limit ) return [self._row_to_event(row) for row in rows] async def read_all( self, from_position: int = 0, limit: int = 1000 ) -> List[Event]: """Read all events globally.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ SELECT id, stream_id, event_type, event_data, metadata, version, global_position, created_at FROM events WHERE global_position > $1 ORDER BY global_position LIMIT $2 """, from_position, limit ) return [self._row_to_event(row) for row in rows] async def subscribe( self, subscription_id: str, handler, from_position: int = 0, batch_size: int = 100 ): """Subscribe to all events from a position.""" # Get checkpoint async with self.pool.acquire() as conn: checkpoint = await conn.fetchval( """ SELECT last_position FROM subscription_checkpoints WHERE subscription_id = $1 """, subscription_id ) position = checkpoint or from_position while True: events = await self.read_all(position, batch_size) if not events: await asyncio.sleep(1) # Poll interval continue for event in events: await handler(event) position = event.global_position # Save checkpoint async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO subscription_checkpoints (subscription_id, last_position) VALUES ($1, $2) ON CONFLICT (subscription_id) DO UPDATE SET last_position = $2, updated_at = NOW() """, subscription_id, position ) def _row_to_event(self, row) -> Event: return Event( event_id=row['id'], stream_id=row['stream_id'], event_type=row['event_type'], data=json.loads(row['event_data']), metadata=json.loads(row['metadata']), version=row['version'], global_position=row['global_position'], created_at=row['created_at'] ) class ConcurrencyError(Exception): """Raised when optimistic concurrency check fails.""" pass ``` ### Template 3: EventStoreDB Usage ```python from esdbclient import EventStoreDBClient, NewEvent, StreamState import json # Connect client = EventStoreDBClient(uri="esdb://localhost:2113?tls=false") # Append events def append_events(stream_name: str, events: list, expected_revision=None): new_events = [ NewEvent( type=event['type'], data=json.dumps(event['data']).encode(), metadata=json.dumps(event.get('metadata', {})).encode() ) for event in events ] if expected_revision is None: state = StreamState.ANY elif expected_revision == -1: state = StreamState.NO_STREAM else: state = expected_revision return client.append_to_stream( stream_name=stream_name, events=new_events, current_version=state ) # Read stream def read_stream(stream_name: str, from_revision: int = 0): events = client.get_stream( stream_name=stream_name, stream_position=from_revision ) return [ { 'type': event.type, 'data': json.loads(event.data), 'metadata': json.loads(event.metadata) if event.metadata else {}, 'stream_position': event.stream_position, 'commit_position': event.commit_position } for event in events ] # Subscribe to all async def subscribe_to_all(handler, from_position: int = 0): subscription = client.subscribe_to_all(commit_position=from_position) async for event in subscription: await handler({ 'type': event.type, 'data': json.loads(event.data), 'stream_id': event.stream_name, 'position': event.commit_position }) # Category projection ($ce-Category) def read_category(category: str): """Read all events for a category using system projection.""" return read_stream(f"$ce-{category}") ``` ### Template 4: DynamoDB Event Store ```python import boto3 from boto3.dynamodb.conditions import Key from datetime import datetime import json import uuid class DynamoEventStore: def __init__(self, table_name: str): self.dynamodb = boto3.resource('dynamodb') self.table = self.dynamodb.Table(table_name) def append_events(self, stream_id: str, events: list, expected_version: int = None): """Append events with conditional write for concurrency.""" with self.table.batch_writer() as batch: for i, event in enumerate(events): version = (expected_version or 0) + i + 1 item = { 'PK': f"STREAM#{stream_id}", 'SK': f"VERSION#{version:020d}", 'GSI1PK': 'EVENTS', 'GSI1SK': datetime.utcnow().isoformat(), 'event_id': str(uuid.uuid4()), 'stream_id': stream_id, 'event_type': event['type'], 'event_data': json.dumps(event['data']), 'version': version, 'created_at': datetime.utcnow().isoformat() } batch.put_item(Item=item) return events def read_stream(self, stream_id: str, from_version: int = 0): """Read events from a stream.""" response = self.table.query( KeyConditionExpression=Key('PK').eq(f"STREAM#{stream_id}") & Key('SK').gte(f"VERSION#{from_version:020d}") ) return [ { 'event_type': item['event_type'], 'data': json.loads(item['event_data']), 'version': item['version'] } for item in response['Items'] ] # Table definition (CloudFormation/Terraform) """ DynamoDB Table: - PK (Partition Key): String - SK (Sort Key): String - GSI1PK, GSI1SK for global ordering Capacity: On-demand or provisioned based on throughput needs """ ``` ## Best Practices ### Do's - **Use stream IDs that include aggregate type** - `Order-{uuid}` - **Include correlation/causation IDs** - For tracing - **Version events from day one** - Plan for schema evolution - **Implement idempotency** - Use event IDs for deduplication - **Index appropriately** - For your query patterns ### Don'ts - **Don't update or delete events** - They're immutable facts - **Don't store large payloads** - Keep events small - **Don't skip optimistic concurrency** - Prevents data corruption - **Don't ignore backpressure** - Handle slow consumers ## Resources - [EventStoreDB](https://www.eventstore.com/) - [Marten Events](https://martendb.io/events/) - [Event Sourcing Pattern](https://docs.microsoft.com/en-us/azure/architecture/patterns/event-sourcing)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

projection-patterns

Build read models and projections from event streams. Use when

coding
โญ1
# Projection Patterns Comprehensive guide to building projections and read models for event-sourced systems. ## When to Use This Skill - Building CQRS read models - Creating materialized views from events - Optimizing query performance - Implementing real-time dashboards - Building search indexes from events - Aggregating data across streams ## Core Concepts ### 1. Projection Architecture ``` โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ Event Store โ”‚โ”€โ”€โ”€โ”€โ–บโ”‚ Projector โ”‚โ”€โ”€โ”€โ”€โ–บโ”‚ Read Model โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ (Database) โ”‚ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ โ”‚ โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ โ”‚ โ”‚ Events โ”‚ โ”‚ โ”‚ โ”‚ Handler โ”‚ โ”‚ โ”‚ โ”‚ Tables โ”‚ โ”‚ โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”‚ โ”‚ Logic โ”‚ โ”‚ โ”‚ โ”‚ Views โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ”‚ โ”‚ โ”‚ Cache โ”‚ โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ ``` ### 2. Projection Types | Type | Description | Use Case | | -------------- | --------------------------- | ---------------------- | | **Live** | Real-time from subscription | Current state queries | | **Catchup** | Process historical events | Rebuilding read models | | **Persistent** | Stores checkpoint | Resume after restart | | **Inline** | Same transaction as write | Strong consistency | ## Templates ### Template 1: Basic Projector ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Dict, Any, Callable, List import asyncpg @dataclass class Event: stream_id: str event_type: str data: dict version: int global_position: int class Projection(ABC): """Base class for projections.""" @property @abstractmethod def name(self) -> str: """Unique projection name for checkpointing.""" pass @abstractmethod def handles(self) -> List[str]: """List of event types this projection handles.""" pass @abstractmethod async def apply(self, event: Event) -> None: """Apply event to the read model.""" pass class Projector: """Runs projections from event store.""" def __init__(self, event_store, checkpoint_store): self.event_store = event_store self.checkpoint_store = checkpoint_store self.projections: List[Projection] = [] def register(self, projection: Projection): self.projections.append(projection) async def run(self, batch_size: int = 100): """Run all projections continuously.""" while True: for projection in self.projections: await self._run_projection(projection, batch_size) await asyncio.sleep(0.1) async def _run_projection(self, projection: Projection, batch_size: int): checkpoint = await self.checkpoint_store.get(projection.name) position = checkpoint or 0 events = await self.event_store.read_all(position, batch_size) for event in events: if event.event_type in projection.handles(): await projection.apply(event) await self.checkpoint_store.save( projection.name, event.global_position ) async def rebuild(self, projection: Projection): """Rebuild a projection from scratch.""" await self.checkpoint_store.delete(projection.name) # Optionally clear read model tables await self._run_projection(projection, batch_size=1000) ``` ### Template 2: Order Summary Projection ```python class OrderSummaryProjection(Projection): """Projects order events to a summary read model.""" def __init__(self, db_pool: asyncpg.Pool): self.pool = db_pool @property def name(self) -> str: return "order_summary" def handles(self) -> List[str]: return [ "OrderCreated", "OrderItemAdded", "OrderItemRemoved", "OrderShipped", "OrderCompleted", "OrderCancelled" ] async def apply(self, event: Event) -> None: handlers = { "OrderCreated": self._handle_created, "OrderItemAdded": self._handle_item_added, "OrderItemRemoved": self._handle_item_removed, "OrderShipped": self._handle_shipped, "OrderCompleted": self._handle_completed, "OrderCancelled": self._handle_cancelled, } handler = handlers.get(event.event_type) if handler: await handler(event) async def _handle_created(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO order_summaries (order_id, customer_id, status, total_amount, item_count, created_at) VALUES ($1, $2, $3, $4, $5, $6) """, event.data['order_id'], event.data['customer_id'], 'pending', 0, 0, event.data['created_at'] ) async def _handle_item_added(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET total_amount = total_amount + $2, item_count = item_count + 1, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['price'] * event.data['quantity'] ) async def _handle_item_removed(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET total_amount = total_amount - $2, item_count = item_count - 1, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['price'] * event.data['quantity'] ) async def _handle_shipped(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET status = 'shipped', shipped_at = $2, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['shipped_at'] ) async def _handle_completed(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET status = 'completed', completed_at = $2, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['completed_at'] ) async def _handle_cancelled(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET status = 'cancelled', cancelled_at = $2, cancellation_reason = $3, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['cancelled_at'], event.data.get('reason') ) ``` ### Template 3: Elasticsearch Search Projection ```python from elasticsearch import AsyncElasticsearch class ProductSearchProjection(Projection): """Projects product events to Elasticsearch for full-text search.""" def __init__(self, es_client: AsyncElasticsearch): self.es = es_client self.index = "products" @property def name(self) -> str: return "product_search" def handles(self) -> List[str]: return [ "ProductCreated", "ProductUpdated", "ProductPriceChanged", "ProductDeleted" ] async def apply(self, event: Event) -> None: if event.event_type == "ProductCreated": await self.es.index( index=self.index, id=event.data['product_id'], document={ 'name': event.data['name'], 'description': event.data['description'], 'category': event.data['category'], 'price': event.data['price'], 'tags': event.data.get('tags', []), 'created_at': event.data['created_at'] } ) elif event.event_type == "ProductUpdated": await self.es.update( index=self.index, id=event.data['product_id'], doc={ 'name': event.data['name'], 'description': event.data['description'], 'category': event.data['category'], 'tags': event.data.get('tags', []), 'updated_at': event.data['updated_at'] } ) elif event.event_type == "ProductPriceChanged": await self.es.update( index=self.index, id=event.data['product_id'], doc={ 'price': event.data['new_price'], 'price_updated_at': event.data['changed_at'] } ) elif event.event_type == "ProductDeleted": await self.es.delete( index=self.index, id=event.data['product_id'] ) ``` ### Template 4: Aggregating Projection ```python class DailySalesProjection(Projection): """Aggregates sales data by day for reporting.""" def __init__(self, db_pool: asyncpg.Pool): self.pool = db_pool @property def name(self) -> str: return "daily_sales" def handles(self) -> List[str]: return ["OrderCompleted", "OrderRefunded"] async def apply(self, event: Event) -> None: if event.event_type == "OrderCompleted": await self._increment_sales(event) elif event.event_type == "OrderRefunded": await self._decrement_sales(event) async def _increment_sales(self, event: Event): date = event.data['completed_at'][:10] # YYYY-MM-DD async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO daily_sales (date, total_orders, total_revenue, total_items) VALUES ($1, 1, $2, $3) ON CONFLICT (date) DO UPDATE SET total_orders = daily_sales.total_orders + 1, total_revenue = daily_sales.total_revenue + $2, total_items = daily_sales.total_items + $3, updated_at = NOW() """, date, event.data['total_amount'], event.data['item_count'] ) async def _decrement_sales(self, event: Event): date = event.data['original_completed_at'][:10] async with self.pool.acquire() as conn: await conn.execute( """ UPDATE daily_sales SET total_orders = total_orders - 1, total_revenue = total_revenue - $2, total_refunds = total_refunds + $2, updated_at = NOW() WHERE date = $1 """, date, event.data['refund_amount'] ) ``` ### Template 5: Multi-Table Projection ```python class CustomerActivityProjection(Projection): """Projects customer activity across multiple tables.""" def __init__(self, db_pool: asyncpg.Pool): self.pool = db_pool @property def name(self) -> str: return "customer_activity" def handles(self) -> List[str]: return [ "CustomerCreated", "OrderCompleted", "ReviewSubmitted", "CustomerTierChanged" ] async def apply(self, event: Event) -> None: async with self.pool.acquire() as conn: async with conn.transaction(): if event.event_type == "CustomerCreated": # Insert into customers table await conn.execute( """ INSERT INTO customers (customer_id, email, name, tier, created_at) VALUES ($1, $2, $3, 'bronze', $4) """, event.data['customer_id'], event.data['email'], event.data['name'], event.data['created_at'] ) # Initialize activity summary await conn.execute( """ INSERT INTO customer_activity_summary (customer_id, total_orders, total_spent, total_reviews) VALUES ($1, 0, 0, 0) """, event.data['customer_id'] ) elif event.event_type == "OrderCompleted": # Update activity summary await conn.execute( """ UPDATE customer_activity_summary SET total_orders = total_orders + 1, total_spent = total_spent + $2, last_order_at = $3 WHERE customer_id = $1 """, event.data['customer_id'], event.data['total_amount'], event.data['completed_at'] ) # Insert into order history await conn.execute( """ INSERT INTO customer_order_history (customer_id, order_id, amount, completed_at) VALUES ($1, $2, $3, $4) """, event.data['customer_id'], event.data['order_id'], event.data['total_amount'], event.data['completed_at'] ) elif event.event_type == "ReviewSubmitted": await conn.execute( """ UPDATE customer_activity_summary SET total_reviews = total_reviews + 1, last_review_at = $2 WHERE customer_id = $1 """, event.data['customer_id'], event.data['submitted_at'] ) elif event.event_type == "CustomerTierChanged": await conn.execute( """ UPDATE customers SET tier = $2, updated_at = NOW() WHERE customer_id = $1 """, event.data['customer_id'], event.data['new_tier'] ) ``` ## Best Practices ### Do's - **Make projections idempotent** - Safe to replay - **Use transactions** - For multi-table updates - **Store checkpoints** - Resume after failures - **Monitor lag** - Alert on projection delays - **Plan for rebuilds** - Design for reconstruction ### Don'ts - **Don't couple projections** - Each is independent - **Don't skip error handling** - Log and alert on failures - **Don't ignore ordering** - Events must be processed in order - **Don't over-normalize** - Denormalize for query patterns ## Resources - [CQRS Pattern](https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs) - [Projection Building Blocks](https://zimarev.com/blog/event-sourcing/projections/)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

workflow-orchestration-patterns

Design durable workflows with Temporal for distributed systems.

coding
โญ1
# Workflow Orchestration Patterns Master workflow orchestration architecture with Temporal, covering fundamental design decisions, resilience patterns, and best practices for building reliable distributed systems. ## When to Use Workflow Orchestration ### Ideal Use Cases (Source: docs.temporal.io) - **Multi-step processes** spanning machines/services/databases - **Distributed transactions** requiring all-or-nothing semantics - **Long-running workflows** (hours to years) with automatic state persistence - **Failure recovery** that must resume from last successful step - **Business processes**: bookings, orders, campaigns, approvals - **Entity lifecycle management**: inventory tracking, account management, cart workflows - **Infrastructure automation**: CI/CD pipelines, provisioning, deployments - **Human-in-the-loop** systems requiring timeouts and escalations ### When NOT to Use - Simple CRUD operations (use direct API calls) - Pure data processing pipelines (use Airflow, batch processing) - Stateless request/response (use standard APIs) - Real-time streaming (use Kafka, event processors) ## Critical Design Decision: Workflows vs Activities **The Fundamental Rule** (Source: temporal.io/blog/workflow-engine-principles): - **Workflows** = Orchestration logic and decision-making - **Activities** = External interactions (APIs, databases, network calls) ### Workflows (Orchestration) **Characteristics:** - Contain business logic and coordination - **MUST be deterministic** (same inputs โ†’ same outputs) - **Cannot** perform direct external calls - State automatically preserved across failures - Can run for years despite infrastructure failures **Example workflow tasks:** - Decide which steps to execute - Handle compensation logic - Manage timeouts and retries - Coordinate child workflows ### Activities (External Interactions) **Characteristics:** - Handle all external system interactions - Can be non-deterministic (API calls, DB writes) - Include built-in timeouts and retry logic - **Must be idempotent** (calling N times = calling once) - Short-lived (seconds to minutes typically) **Example activity tasks:** - Call payment gateway API - Write to database - Send emails or notifications - Query external services ### Design Decision Framework ``` Does it touch external systems? โ†’ Activity Is it orchestration/decision logic? โ†’ Workflow ``` ## Core Workflow Patterns ### 1. Saga Pattern with Compensation **Purpose**: Implement distributed transactions with rollback capability **Pattern** (Source: temporal.io/blog/compensating-actions-part-of-a-complete-breakfast-with-sagas): ``` For each step: 1. Register compensation BEFORE executing 2. Execute the step (via activity) 3. On failure, run all compensations in reverse order (LIFO) ``` **Example: Payment Workflow** 1. Reserve inventory (compensation: release inventory) 2. Charge payment (compensation: refund payment) 3. Fulfill order (compensation: cancel fulfillment) **Critical Requirements:** - Compensations must be idempotent - Register compensation BEFORE executing step - Run compensations in reverse order - Handle partial failures gracefully ### 2. Entity Workflows (Actor Model) **Purpose**: Long-lived workflow representing single entity instance **Pattern** (Source: docs.temporal.io/evaluate/use-cases-design-patterns): - One workflow execution = one entity (cart, account, inventory item) - Workflow persists for entity lifetime - Receives signals for state changes - Supports queries for current state **Example Use Cases:** - Shopping cart (add items, checkout, expiration) - Bank account (deposits, withdrawals, balance checks) - Product inventory (stock updates, reservations) **Benefits:** - Encapsulates entity behavior - Guarantees consistency per entity - Natural event sourcing ### 3. Fan-Out/Fan-In (Parallel Execution) **Purpose**: Execute multiple tasks in parallel, aggregate results **Pattern:** - Spawn child workflows or parallel activities - Wait for all to complete - Aggregate results - Handle partial failures **Scaling Rule** (Source: temporal.io/blog/workflow-engine-principles): - Don't scale individual workflows - For 1M tasks: spawn 1K child workflows ร— 1K tasks each - Keep each workflow bounded ### 4. Async Callback Pattern **Purpose**: Wait for external event or human approval **Pattern:** - Workflow sends request and waits for signal - External system processes asynchronously - Sends signal to resume workflow - Workflow continues with response **Use Cases:** - Human approval workflows - Webhook callbacks - Long-running external processes ## State Management and Determinism ### Automatic State Preservation **How Temporal Works** (Source: docs.temporal.io/workflows): - Complete program state preserved automatically - Event History records every command and event - Seamless recovery from crashes - Applications restore pre-failure state ### Determinism Constraints **Workflows Execute as State Machines**: - Replay behavior must be consistent - Same inputs โ†’ identical outputs every time **Prohibited in Workflows** (Source: docs.temporal.io/workflows): - โŒ Threading, locks, synchronization primitives - โŒ Random number generation (`random()`) - โŒ Global state or static variables - โŒ System time (`datetime.now()`) - โŒ Direct file I/O or network calls - โŒ Non-deterministic libraries **Allowed in Workflows**: - โœ… `workflow.now()` (deterministic time) - โœ… `workflow.random()` (deterministic random) - โœ… Pure functions and calculations - โœ… Calling activities (non-deterministic operations) ### Versioning Strategies **Challenge**: Changing workflow code while old executions still running **Solutions**: 1. **Versioning API**: Use `workflow.get_version()` for safe changes 2. **New Workflow Type**: Create new workflow, route new executions to it 3. **Backward Compatibility**: Ensure old events replay correctly ## Resilience and Error Handling ### Retry Policies **Default Behavior**: Temporal retries activities forever **Configure Retry**: - Initial retry interval - Backoff coefficient (exponential backoff) - Maximum interval (cap retry delay) - Maximum attempts (eventually fail) **Non-Retryable Errors**: - Invalid input (validation failures) - Business rule violations - Permanent failures (resource not found) ### Idempotency Requirements **Why Critical** (Source: docs.temporal.io/activities): - Activities may execute multiple times - Network failures trigger retries - Duplicate execution must be safe **Implementation Strategies**: - Idempotency keys (deduplication) - Check-then-act with unique constraints - Upsert operations instead of insert - Track processed request IDs ### Activity Heartbeats **Purpose**: Detect stalled long-running activities **Pattern**: - Activity sends periodic heartbeat - Includes progress information - Timeout if no heartbeat received - Enables progress-based retry ## Best Practices ### Workflow Design 1. **Keep workflows focused** - Single responsibility per workflow 2. **Small workflows** - Use child workflows for scalability 3. **Clear boundaries** - Workflow orchestrates, activities execute 4. **Test locally** - Use time-skipping test environment ### Activity Design 1. **Idempotent operations** - Safe to retry 2. **Short-lived** - Seconds to minutes, not hours 3. **Timeout configuration** - Always set timeouts 4. **Heartbeat for long tasks** - Report progress 5. **Error handling** - Distinguish retryable vs non-retryable ### Common Pitfalls **Workflow Violations**: - Using `datetime.now()` instead of `workflow.now()` - Threading or async operations in workflow code - Calling external APIs directly from workflow - Non-deterministic logic in workflows **Activity Mistakes**: - Non-idempotent operations (can't handle retries) - Missing timeouts (activities run forever) - No error classification (retry validation errors) - Ignoring payload limits (2MB per argument) ### Operational Considerations **Monitoring**: - Workflow execution duration - Activity failure rates - Retry attempts and backoff - Pending workflow counts **Scalability**: - Horizontal scaling with workers - Task queue partitioning - Child workflow decomposition - Activity batching when appropriate ## Additional Resources **Official Documentation**: - Temporal Core Concepts: docs.temporal.io/workflows - Workflow Patterns: docs.temporal.io/evaluate/use-cases-design-patterns - Best Practices: docs.temporal.io/develop/best-practices - Saga Pattern: temporal.io/blog/saga-pattern-made-easy **Key Principles**: 1. Workflows = orchestration, Activities = external calls 2. Determinism is non-negotiable for workflows 3. Idempotency is critical for activities 4. State preservation is automatic 5. Design for failure and recovery
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

nft-standards

Implement NFT standards (ERC-721, ERC-1155) with proper metadata

coding
โญ1
# NFT Standards Master ERC-721 and ERC-1155 NFT standards, metadata best practices, and advanced NFT features. ## When to Use This Skill - Creating NFT collections (art, gaming, collectibles) - Implementing marketplace functionality - Building on-chain or off-chain metadata - Creating soulbound tokens (non-transferable) - Implementing royalties and revenue sharing - Developing dynamic/evolving NFTs ## ERC-721 (Non-Fungible Token Standard) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC721/extensions/ERC721URIStorage.sol"; import "@openzeppelin/contracts/token/ERC721/extensions/ERC721Enumerable.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; import "@openzeppelin/contracts/utils/Counters.sol"; contract MyNFT is ERC721URIStorage, ERC721Enumerable, Ownable { using Counters for Counters.Counter; Counters.Counter private _tokenIds; uint256 public constant MAX_SUPPLY = 10000; uint256 public constant MINT_PRICE = 0.08 ether; uint256 public constant MAX_PER_MINT = 20; constructor() ERC721("MyNFT", "MNFT") {} function mint(uint256 quantity) external payable { require(quantity > 0 && quantity <= MAX_PER_MINT, "Invalid quantity"); require(_tokenIds.current() + quantity <= MAX_SUPPLY, "Exceeds max supply"); require(msg.value >= MINT_PRICE * quantity, "Insufficient payment"); for (uint256 i = 0; i < quantity; i++) { _tokenIds.increment(); uint256 newTokenId = _tokenIds.current(); _safeMint(msg.sender, newTokenId); _setTokenURI(newTokenId, generateTokenURI(newTokenId)); } } function generateTokenURI(uint256 tokenId) internal pure returns (string memory) { // Return IPFS URI or on-chain metadata return string(abi.encodePacked("ipfs://QmHash/", Strings.toString(tokenId), ".json")); } // Required overrides function _beforeTokenTransfer( address from, address to, uint256 tokenId, uint256 batchSize ) internal override(ERC721, ERC721Enumerable) { super._beforeTokenTransfer(from, to, tokenId, batchSize); } function _burn(uint256 tokenId) internal override(ERC721, ERC721URIStorage) { super._burn(tokenId); } function tokenURI(uint256 tokenId) public view override(ERC721, ERC721URIStorage) returns (string memory) { return super.tokenURI(tokenId); } function supportsInterface(bytes4 interfaceId) public view override(ERC721, ERC721Enumerable) returns (bool) { return super.supportsInterface(interfaceId); } function withdraw() external onlyOwner { payable(owner()).transfer(address(this).balance); } } ``` ## ERC-1155 (Multi-Token Standard) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC1155/ERC1155.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract GameItems is ERC1155, Ownable { uint256 public constant SWORD = 1; uint256 public constant SHIELD = 2; uint256 public constant POTION = 3; mapping(uint256 => uint256) public tokenSupply; mapping(uint256 => uint256) public maxSupply; constructor() ERC1155("ipfs://QmBaseHash/{id}.json") { maxSupply[SWORD] = 1000; maxSupply[SHIELD] = 500; maxSupply[POTION] = 10000; } function mint( address to, uint256 id, uint256 amount ) external onlyOwner { require(tokenSupply[id] + amount <= maxSupply[id], "Exceeds max supply"); _mint(to, id, amount, ""); tokenSupply[id] += amount; } function mintBatch( address to, uint256[] memory ids, uint256[] memory amounts ) external onlyOwner { for (uint256 i = 0; i < ids.length; i++) { require(tokenSupply[ids[i]] + amounts[i] <= maxSupply[ids[i]], "Exceeds max supply"); tokenSupply[ids[i]] += amounts[i]; } _mintBatch(to, ids, amounts, ""); } function burn( address from, uint256 id, uint256 amount ) external { require(from == msg.sender || isApprovedForAll(from, msg.sender), "Not authorized"); _burn(from, id, amount); tokenSupply[id] -= amount; } } ``` ## Metadata Standards ### Off-Chain Metadata (IPFS) ```json { "name": "NFT #1", "description": "Description of the NFT", "image": "ipfs://QmImageHash", "attributes": [ { "trait_type": "Background", "value": "Blue" }, { "trait_type": "Rarity", "value": "Legendary" }, { "trait_type": "Power", "value": 95, "display_type": "number", "max_value": 100 } ] } ``` ### On-Chain Metadata ```solidity contract OnChainNFT is ERC721 { struct Traits { uint8 background; uint8 body; uint8 head; uint8 rarity; } mapping(uint256 => Traits) public tokenTraits; function tokenURI(uint256 tokenId) public view override returns (string memory) { Traits memory traits = tokenTraits[tokenId]; string memory json = Base64.encode( bytes( string( abi.encodePacked( '{"name": "NFT #', Strings.toString(tokenId), '",', '"description": "On-chain NFT",', '"image": "data:image/svg+xml;base64,', generateSVG(traits), '",', '"attributes": [', '{"trait_type": "Background", "value": "', Strings.toString(traits.background), '"},', '{"trait_type": "Rarity", "value": "', getRarityName(traits.rarity), '"}', ']}' ) ) ) ); return string(abi.encodePacked("data:application/json;base64,", json)); } function generateSVG(Traits memory traits) internal pure returns (string memory) { // Generate SVG based on traits return "..."; } } ``` ## Royalties (EIP-2981) ```solidity import "@openzeppelin/contracts/interfaces/IERC2981.sol"; contract NFTWithRoyalties is ERC721, IERC2981 { address public royaltyRecipient; uint96 public royaltyFee = 500; // 5% constructor() ERC721("Royalty NFT", "RNFT") { royaltyRecipient = msg.sender; } function royaltyInfo(uint256 tokenId, uint256 salePrice) external view override returns (address receiver, uint256 royaltyAmount) { return (royaltyRecipient, (salePrice * royaltyFee) / 10000); } function setRoyalty(address recipient, uint96 fee) external onlyOwner { require(fee <= 1000, "Royalty fee too high"); // Max 10% royaltyRecipient = recipient; royaltyFee = fee; } function supportsInterface(bytes4 interfaceId) public view override(ERC721, IERC165) returns (bool) { return interfaceId == type(IERC2981).interfaceId || super.supportsInterface(interfaceId); } } ``` ## Soulbound Tokens (Non-Transferable) ```solidity contract SoulboundToken is ERC721 { constructor() ERC721("Soulbound", "SBT") {} function _beforeTokenTransfer( address from, address to, uint256 tokenId, uint256 batchSize ) internal virtual override { require(from == address(0) || to == address(0), "Token is soulbound"); super._beforeTokenTransfer(from, to, tokenId, batchSize); } function mint(address to) external { uint256 tokenId = totalSupply() + 1; _safeMint(to, tokenId); } // Burn is allowed (user can destroy their SBT) function burn(uint256 tokenId) external { require(ownerOf(tokenId) == msg.sender, "Not token owner"); _burn(tokenId); } } ``` ## Dynamic NFTs ```solidity contract DynamicNFT is ERC721 { struct TokenState { uint256 level; uint256 experience; uint256 lastUpdated; } mapping(uint256 => TokenState) public tokenStates; function gainExperience(uint256 tokenId, uint256 exp) external { require(ownerOf(tokenId) == msg.sender, "Not token owner"); TokenState storage state = tokenStates[tokenId]; state.experience += exp; // Level up logic if (state.experience >= state.level * 100) { state.level++; } state.lastUpdated = block.timestamp; } function tokenURI(uint256 tokenId) public view override returns (string memory) { TokenState memory state = tokenStates[tokenId]; // Generate metadata based on current state return generateMetadata(tokenId, state); } function generateMetadata(uint256 tokenId, TokenState memory state) internal pure returns (string memory) { // Dynamic metadata generation return ""; } } ``` ## Gas-Optimized Minting (ERC721A) ```solidity import "erc721a/contracts/ERC721A.sol"; contract OptimizedNFT is ERC721A { uint256 public constant MAX_SUPPLY = 10000; uint256 public constant MINT_PRICE = 0.05 ether; constructor() ERC721A("Optimized NFT", "ONFT") {} function mint(uint256 quantity) external payable { require(_totalMinted() + quantity <= MAX_SUPPLY, "Exceeds max supply"); require(msg.value >= MINT_PRICE * quantity, "Insufficient payment"); _mint(msg.sender, quantity); } function _baseURI() internal pure override returns (string memory) { return "ipfs://QmBaseHash/"; } } ``` ## Resources - **references/erc721.md**: ERC-721 specification details - **references/erc1155.md**: ERC-1155 multi-token standard - **references/metadata-standards.md**: Metadata best practices - **references/enumeration.md**: Token enumeration patterns - **assets/erc721-contract.sol**: Production ERC-721 template - **assets/erc1155-contract.sol**: Production ERC-1155 template - **assets/metadata-schema.json**: Standard metadata format - **assets/metadata-uploader.py**: IPFS upload utility ## Best Practices 1. **Use OpenZeppelin**: Battle-tested implementations 2. **Pin Metadata**: Use IPFS with pinning service 3. **Implement Royalties**: EIP-2981 for marketplace compatibility 4. **Gas Optimization**: Use ERC721A for batch minting 5. **Reveal Mechanism**: Placeholder โ†’ reveal pattern 6. **Enumeration**: Support walletOfOwner for marketplaces 7. **Whitelist**: Merkle trees for efficient whitelisting ## Marketplace Integration - OpenSea: ERC-721/1155, metadata standards - LooksRare: Royalty enforcement - Rarible: Protocol fees, lazy minting - Blur: Gas-optimized trading
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

solidity-security

Master smart contract security best practices to prevent common

security
โญ1
# Solidity Security Master smart contract security best practices, vulnerability prevention, and secure Solidity development patterns. ## When to Use This Skill - Writing secure smart contracts - Auditing existing contracts for vulnerabilities - Implementing secure DeFi protocols - Preventing reentrancy, overflow, and access control issues - Optimizing gas usage while maintaining security - Preparing contracts for professional audits - Understanding common attack vectors ## Critical Vulnerabilities ### 1. Reentrancy Attacker calls back into your contract before state is updated. **Vulnerable Code:** ```solidity // VULNERABLE TO REENTRANCY contract VulnerableBank { mapping(address => uint256) public balances; function withdraw() public { uint256 amount = balances[msg.sender]; // DANGER: External call before state update (bool success, ) = msg.sender.call{value: amount}(""); require(success); balances[msg.sender] = 0; // Too late! } } ``` **Secure Pattern (Checks-Effects-Interactions):** ```solidity contract SecureBank { mapping(address => uint256) public balances; function withdraw() public { uint256 amount = balances[msg.sender]; require(amount > 0, "Insufficient balance"); // EFFECTS: Update state BEFORE external call balances[msg.sender] = 0; // INTERACTIONS: External call last (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` **Alternative: ReentrancyGuard** ```solidity import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; contract SecureBank is ReentrancyGuard { mapping(address => uint256) public balances; function withdraw() public nonReentrant { uint256 amount = balances[msg.sender]; require(amount > 0, "Insufficient balance"); balances[msg.sender] = 0; (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` ### 2. Integer Overflow/Underflow **Vulnerable Code (Solidity < 0.8.0):** ```solidity // VULNERABLE contract VulnerableToken { mapping(address => uint256) public balances; function transfer(address to, uint256 amount) public { // No overflow check - can wrap around balances[msg.sender] -= amount; // Can underflow! balances[to] += amount; // Can overflow! } } ``` **Secure Pattern (Solidity >= 0.8.0):** ```solidity // Solidity 0.8+ has built-in overflow/underflow checks contract SecureToken { mapping(address => uint256) public balances; function transfer(address to, uint256 amount) public { // Automatically reverts on overflow/underflow balances[msg.sender] -= amount; balances[to] += amount; } } ``` **For Solidity < 0.8.0, use SafeMath:** ```solidity import "@openzeppelin/contracts/utils/math/SafeMath.sol"; contract SecureToken { using SafeMath for uint256; mapping(address => uint256) public balances; function transfer(address to, uint256 amount) public { balances[msg.sender] = balances[msg.sender].sub(amount); balances[to] = balances[to].add(amount); } } ``` ### 3. Access Control **Vulnerable Code:** ```solidity // VULNERABLE: Anyone can call critical functions contract VulnerableContract { address public owner; function withdraw(uint256 amount) public { // No access control! payable(msg.sender).transfer(amount); } } ``` **Secure Pattern:** ```solidity import "@openzeppelin/contracts/access/Ownable.sol"; contract SecureContract is Ownable { function withdraw(uint256 amount) public onlyOwner { payable(owner()).transfer(amount); } } // Or implement custom role-based access contract RoleBasedContract { mapping(address => bool) public admins; modifier onlyAdmin() { require(admins[msg.sender], "Not an admin"); _; } function criticalFunction() public onlyAdmin { // Protected function } } ``` ### 4. Front-Running **Vulnerable:** ```solidity // VULNERABLE TO FRONT-RUNNING contract VulnerableDEX { function swap(uint256 amount, uint256 minOutput) public { // Attacker sees this in mempool and front-runs uint256 output = calculateOutput(amount); require(output >= minOutput, "Slippage too high"); // Perform swap } } ``` **Mitigation:** ```solidity contract SecureDEX { mapping(bytes32 => bool) public usedCommitments; // Step 1: Commit to trade function commitTrade(bytes32 commitment) public { usedCommitments[commitment] = true; } // Step 2: Reveal trade (next block) function revealTrade( uint256 amount, uint256 minOutput, bytes32 secret ) public { bytes32 commitment = keccak256(abi.encodePacked( msg.sender, amount, minOutput, secret )); require(usedCommitments[commitment], "Invalid commitment"); // Perform swap } } ``` ## Security Best Practices ### Checks-Effects-Interactions Pattern ```solidity contract SecurePattern { mapping(address => uint256) public balances; function withdraw(uint256 amount) public { // 1. CHECKS: Validate conditions require(amount <= balances[msg.sender], "Insufficient balance"); require(amount > 0, "Amount must be positive"); // 2. EFFECTS: Update state balances[msg.sender] -= amount; // 3. INTERACTIONS: External calls last (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` ### Pull Over Push Pattern ```solidity // Prefer this (pull) contract SecurePayment { mapping(address => uint256) public pendingWithdrawals; function recordPayment(address recipient, uint256 amount) internal { pendingWithdrawals[recipient] += amount; } function withdraw() public { uint256 amount = pendingWithdrawals[msg.sender]; require(amount > 0, "Nothing to withdraw"); pendingWithdrawals[msg.sender] = 0; payable(msg.sender).transfer(amount); } } // Over this (push) contract RiskyPayment { function distributePayments(address[] memory recipients, uint256[] memory amounts) public { for (uint i = 0; i < recipients.length; i++) { // If any transfer fails, entire batch fails payable(recipients[i]).transfer(amounts[i]); } } } ``` ### Input Validation ```solidity contract SecureContract { function transfer(address to, uint256 amount) public { // Validate inputs require(to != address(0), "Invalid recipient"); require(to != address(this), "Cannot send to contract"); require(amount > 0, "Amount must be positive"); require(amount <= balances[msg.sender], "Insufficient balance"); // Proceed with transfer balances[msg.sender] -= amount; balances[to] += amount; } } ``` ### Emergency Stop (Circuit Breaker) ```solidity import "@openzeppelin/contracts/security/Pausable.sol"; contract EmergencyStop is Pausable, Ownable { function criticalFunction() public whenNotPaused { // Function logic } function emergencyStop() public onlyOwner { _pause(); } function resume() public onlyOwner { _unpause(); } } ``` ## Gas Optimization ### Use `uint256` Instead of Smaller Types ```solidity // More gas efficient contract GasEfficient { uint256 public value; // Optimal function set(uint256 _value) public { value = _value; } } // Less efficient contract GasInefficient { uint8 public value; // Still uses 256-bit slot function set(uint8 _value) public { value = _value; // Extra gas for type conversion } } ``` ### Pack Storage Variables ```solidity // Gas efficient (3 variables in 1 slot) contract PackedStorage { uint128 public a; // Slot 0 uint64 public b; // Slot 0 uint64 public c; // Slot 0 uint256 public d; // Slot 1 } // Gas inefficient (each variable in separate slot) contract UnpackedStorage { uint256 public a; // Slot 0 uint256 public b; // Slot 1 uint256 public c; // Slot 2 uint256 public d; // Slot 3 } ``` ### Use `calldata` Instead of `memory` for Function Arguments ```solidity contract GasOptimized { // More gas efficient function processData(uint256[] calldata data) public pure returns (uint256) { return data[0]; } // Less efficient function processDataMemory(uint256[] memory data) public pure returns (uint256) { return data[0]; } } ``` ### Use Events for Data Storage (When Appropriate) ```solidity contract EventStorage { // Emitting events is cheaper than storage event DataStored(address indexed user, uint256 indexed id, bytes data); function storeData(uint256 id, bytes calldata data) public { emit DataStored(msg.sender, id, data); // Don't store in contract storage unless needed } } ``` ## Common Vulnerabilities Checklist ```solidity // Security Checklist Contract contract SecurityChecklist { /** * [ ] Reentrancy protection (ReentrancyGuard or CEI pattern) * [ ] Integer overflow/underflow (Solidity 0.8+ or SafeMath) * [ ] Access control (Ownable, roles, modifiers) * [ ] Input validation (require statements) * [ ] Front-running mitigation (commit-reveal if applicable) * [ ] Gas optimization (packed storage, calldata) * [ ] Emergency stop mechanism (Pausable) * [ ] Pull over push pattern for payments * [ ] No delegatecall to untrusted contracts * [ ] No tx.origin for authentication (use msg.sender) * [ ] Proper event emission * [ ] External calls at end of function * [ ] Check return values of external calls * [ ] No hardcoded addresses * [ ] Upgrade mechanism (if proxy pattern) */ } ``` ## Testing for Security ```javascript // Hardhat test example const { expect } = require("chai"); const { ethers } = require("hardhat"); describe("Security Tests", function () { it("Should prevent reentrancy attack", async function () { const [attacker] = await ethers.getSigners(); const VictimBank = await ethers.getContractFactory("SecureBank"); const bank = await VictimBank.deploy(); const Attacker = await ethers.getContractFactory("ReentrancyAttacker"); const attackerContract = await Attacker.deploy(bank.address); // Deposit funds await bank.deposit({ value: ethers.utils.parseEther("10") }); // Attempt reentrancy attack await expect( attackerContract.attack({ value: ethers.utils.parseEther("1") }), ).to.be.revertedWith("ReentrancyGuard: reentrant call"); }); it("Should prevent integer overflow", async function () { const Token = await ethers.getContractFactory("SecureToken"); const token = await Token.deploy(); // Attempt overflow await expect(token.transfer(attacker.address, ethers.constants.MaxUint256)) .to.be.reverted; }); it("Should enforce access control", async function () { const [owner, attacker] = await ethers.getSigners(); const Contract = await ethers.getContractFactory("SecureContract"); const contract = await Contract.deploy(); // Attempt unauthorized withdrawal await expect(contract.connect(attacker).withdraw(100)).to.be.revertedWith( "Ownable: caller is not the owner", ); }); }); ``` ## Audit Preparation ```solidity contract WellDocumentedContract { /** * @title Well Documented Contract * @dev Example of proper documentation for audits * @notice This contract handles user deposits and withdrawals */ /// @notice Mapping of user balances mapping(address => uint256) public balances; /** * @dev Deposits ETH into the contract * @notice Anyone can deposit funds */ function deposit() public payable { require(msg.value > 0, "Must send ETH"); balances[msg.sender] += msg.value; } /** * @dev Withdraws user's balance * @notice Follows CEI pattern to prevent reentrancy * @param amount Amount to withdraw in wei */ function withdraw(uint256 amount) public { // CHECKS require(amount <= balances[msg.sender], "Insufficient balance"); // EFFECTS balances[msg.sender] -= amount; // INTERACTIONS (bool success, ) = msg.sender.call{value: amount}(""); require(success, "Transfer failed"); } } ``` ## Resources - **references/reentrancy.md**: Comprehensive reentrancy prevention - **references/access-control.md**: Role-based access patterns - **references/overflow-underflow.md**: SafeMath and integer safety - **references/gas-optimization.md**: Gas saving techniques - **references/vulnerability-patterns.md**: Common vulnerability catalog - **assets/solidity-contracts-templates.sol**: Secure contract templates - **assets/security-checklist.md**: Pre-audit checklist - **scripts/analyze-contract.sh**: Static analysis tools ## Tools for Security Analysis - **Slither**: Static analysis tool - **Mythril**: Security analysis tool - **Echidna**: Fuzzing tool - **Manticore**: Symbolic execution - **Securify**: Automated security scanner ## Common Pitfalls 1. **Using `tx.origin` for Authentication**: Use `msg.sender` instead 2. **Unchecked External Calls**: Always check return values 3. **Delegatecall to Untrusted Contracts**: Can hijack your contract 4. **Floating Pragma**: Pin to specific Solidity version 5. **Missing Events**: Emit events for state changes 6. **Excessive Gas in Loops**: Can hit block gas limit 7. **No Upgrade Path**: Consider proxy patterns if upgrades needed
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

cost-optimization

Optimize cloud costs through resource rightsizing, tagging

architecture
โญ1
# Cloud Cost Optimization Strategies and patterns for optimizing cloud costs across AWS, Azure, and GCP. ## Purpose Implement systematic cost optimization strategies to reduce cloud spending while maintaining performance and reliability. ## When to Use - Reduce cloud spending - Right-size resources - Implement cost governance - Optimize multi-cloud costs - Meet budget constraints ## Cost Optimization Framework ### 1. Visibility - Implement cost allocation tags - Use cloud cost management tools - Set up budget alerts - Create cost dashboards ### 2. Right-Sizing - Analyze resource utilization - Downsize over-provisioned resources - Use auto-scaling - Remove idle resources ### 3. Pricing Models - Use reserved capacity - Leverage spot/preemptible instances - Implement savings plans - Use committed use discounts ### 4. Architecture Optimization - Use managed services - Implement caching - Optimize data transfer - Use lifecycle policies ## AWS Cost Optimization ### Reserved Instances ``` Savings: 30-72% vs On-Demand Term: 1 or 3 years Payment: All/Partial/No upfront Flexibility: Standard or Convertible ``` ### Savings Plans ``` Compute Savings Plans: 66% savings EC2 Instance Savings Plans: 72% savings Applies to: EC2, Fargate, Lambda Flexible across: Instance families, regions, OS ``` ### Spot Instances ``` Savings: Up to 90% vs On-Demand Best for: Batch jobs, CI/CD, stateless workloads Risk: 2-minute interruption notice Strategy: Mix with On-Demand for resilience ``` ### S3 Cost Optimization ```hcl resource "aws_s3_bucket_lifecycle_configuration" "example" { bucket = aws_s3_bucket.example.id rule { id = "transition-to-ia" status = "Enabled" transition { days = 30 storage_class = "STANDARD_IA" } transition { days = 90 storage_class = "GLACIER" } expiration { days = 365 } } } ``` ## Azure Cost Optimization ### Reserved VM Instances - 1 or 3 year terms - Up to 72% savings - Flexible sizing - Exchangeable ### Azure Hybrid Benefit - Use existing Windows Server licenses - Up to 80% savings with RI - Available for Windows and SQL Server ### Azure Advisor Recommendations - Right-size VMs - Delete unused resources - Use reserved capacity - Optimize storage ## GCP Cost Optimization ### Committed Use Discounts - 1 or 3 year commitment - Up to 57% savings - Applies to vCPUs and memory - Resource-based or spend-based ### Sustained Use Discounts - Automatic discounts - Up to 30% for running instances - No commitment required - Applies to Compute Engine, GKE ### Preemptible VMs - Up to 80% savings - 24-hour maximum runtime - Best for batch workloads ## Tagging Strategy ### AWS Tagging ```hcl locals { common_tags = { Environment = "production" Project = "my-project" CostCenter = "engineering" Owner = "team@example.com" ManagedBy = "terraform" } } resource "aws_instance" "example" { ami = "ami-12345678" instance_type = "t3.medium" tags = merge( local.common_tags, { Name = "web-server" } ) } ``` **Reference:** See `references/tagging-standards.md` ## Cost Monitoring ### Budget Alerts ```hcl # AWS Budget resource "aws_budgets_budget" "monthly" { name = "monthly-budget" budget_type = "COST" limit_amount = "1000" limit_unit = "USD" time_period_start = "2024-01-01_00:00" time_unit = "MONTHLY" notification { comparison_operator = "GREATER_THAN" threshold = 80 threshold_type = "PERCENTAGE" notification_type = "ACTUAL" subscriber_email_addresses = ["team@example.com"] } } ``` ### Cost Anomaly Detection - AWS Cost Anomaly Detection - Azure Cost Management alerts - GCP Budget alerts ## Architecture Patterns ### Pattern 1: Serverless First - Use Lambda/Functions for event-driven - Pay only for execution time - Auto-scaling included - No idle costs ### Pattern 2: Right-Sized Databases ``` Development: t3.small RDS Staging: t3.large RDS Production: r6g.2xlarge RDS with read replicas ``` ### Pattern 3: Multi-Tier Storage ``` Hot data: S3 Standard Warm data: S3 Standard-IA (30 days) Cold data: S3 Glacier (90 days) Archive: S3 Deep Archive (365 days) ``` ### Pattern 4: Auto-Scaling ```hcl resource "aws_autoscaling_policy" "scale_up" { name = "scale-up" scaling_adjustment = 2 adjustment_type = "ChangeInCapacity" cooldown = 300 autoscaling_group_name = aws_autoscaling_group.main.name } resource "aws_cloudwatch_metric_alarm" "cpu_high" { alarm_name = "cpu-high" comparison_operator = "GreaterThanThreshold" evaluation_periods = "2" metric_name = "CPUUtilization" namespace = "AWS/EC2" period = "60" statistic = "Average" threshold = "80" alarm_actions = [aws_autoscaling_policy.scale_up.arn] } ``` ## Cost Optimization Checklist - [ ] Implement cost allocation tags - [ ] Delete unused resources (EBS, EIPs, snapshots) - [ ] Right-size instances based on utilization - [ ] Use reserved capacity for steady workloads - [ ] Implement auto-scaling - [ ] Optimize storage classes - [ ] Use lifecycle policies - [ ] Enable cost anomaly detection - [ ] Set budget alerts - [ ] Review costs weekly - [ ] Use spot/preemptible instances - [ ] Optimize data transfer costs - [ ] Implement caching layers - [ ] Use managed services - [ ] Monitor and optimize continuously ## Tools - **AWS:** Cost Explorer, Cost Anomaly Detection, Compute Optimizer - **Azure:** Cost Management, Advisor - **GCP:** Cost Management, Recommender - **Multi-cloud:** CloudHealth, Cloudability, Kubecost ## Reference Files - `references/tagging-standards.md` - Tagging conventions - `assets/cost-analysis-template.xlsx` - Cost analysis spreadsheet ## Related Skills - `terraform-module-library` - For resource provisioning - `multi-cloud-architecture` - For cloud selection
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

service-mesh-observability

Implement comprehensive observability for service meshes including

architecture
โญ1
# Service Mesh Observability Complete guide to observability patterns for Istio, Linkerd, and service mesh deployments. ## When to Use This Skill - Setting up distributed tracing across services - Implementing service mesh metrics and dashboards - Debugging latency and error issues - Defining SLOs for service communication - Visualizing service dependencies - Troubleshooting mesh connectivity ## Core Concepts ### 1. Three Pillars of Observability ``` โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ Observability โ”‚ โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”‚ Metrics โ”‚ Traces โ”‚ Logs โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ โ”‚ โ€ข Request rate โ”‚ โ€ข Span context โ”‚ โ€ข Access logs โ”‚ โ”‚ โ€ข Error rate โ”‚ โ€ข Latency โ”‚ โ€ข Error details โ”‚ โ”‚ โ€ข Latency P50 โ”‚ โ€ข Dependencies โ”‚ โ€ข Debug info โ”‚ โ”‚ โ€ข Saturation โ”‚ โ€ข Bottlenecks โ”‚ โ€ข Audit trail โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ ``` ### 2. Golden Signals for Mesh | Signal | Description | Alert Threshold | | -------------- | ------------------------- | ----------------- | | **Latency** | Request duration P50, P99 | P99 > 500ms | | **Traffic** | Requests per second | Anomaly detection | | **Errors** | 5xx error rate | > 1% | | **Saturation** | Resource utilization | > 80% | ## Templates ### Template 1: Istio with Prometheus & Grafana ```yaml # Install Prometheus apiVersion: v1 kind: ConfigMap metadata: name: prometheus namespace: istio-system data: prometheus.yml: | global: scrape_interval: 15s scrape_configs: - job_name: 'istio-mesh' kubernetes_sd_configs: - role: endpoints namespaces: names: - istio-system relabel_configs: - source_labels: [__meta_kubernetes_service_name] action: keep regex: istio-telemetry --- # ServiceMonitor for Prometheus Operator apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: istio-mesh namespace: istio-system spec: selector: matchLabels: app: istiod endpoints: - port: http-monitoring interval: 15s ``` ### Template 2: Key Istio Metrics Queries ```promql # Request rate by service sum(rate(istio_requests_total{reporter="destination"}[5m])) by (destination_service_name) # Error rate (5xx) sum(rate(istio_requests_total{reporter="destination", response_code=~"5.."}[5m])) / sum(rate(istio_requests_total{reporter="destination"}[5m])) * 100 # P99 latency histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket{reporter="destination"}[5m])) by (le, destination_service_name)) # TCP connections sum(istio_tcp_connections_opened_total{reporter="destination"}) by (destination_service_name) # Request size histogram_quantile(0.99, sum(rate(istio_request_bytes_bucket{reporter="destination"}[5m])) by (le, destination_service_name)) ``` ### Template 3: Jaeger Distributed Tracing ```yaml # Jaeger installation for Istio apiVersion: install.istio.io/v1alpha1 kind: IstioOperator spec: meshConfig: enableTracing: true defaultConfig: tracing: sampling: 100.0 # 100% in dev, lower in prod zipkin: address: jaeger-collector.istio-system:9411 --- # Jaeger deployment apiVersion: apps/v1 kind: Deployment metadata: name: jaeger namespace: istio-system spec: selector: matchLabels: app: jaeger template: metadata: labels: app: jaeger spec: containers: - name: jaeger image: jaegertracing/all-in-one:1.50 ports: - containerPort: 5775 # UDP - containerPort: 6831 # Thrift - containerPort: 6832 # Thrift - containerPort: 5778 # Config - containerPort: 16686 # UI - containerPort: 14268 # HTTP - containerPort: 14250 # gRPC - containerPort: 9411 # Zipkin env: - name: COLLECTOR_ZIPKIN_HOST_PORT value: ":9411" ``` ### Template 4: Linkerd Viz Dashboard ```bash # Install Linkerd viz extension linkerd viz install | kubectl apply -f - # Access dashboard linkerd viz dashboard # CLI commands for observability # Top requests linkerd viz top deploy/my-app # Per-route metrics linkerd viz routes deploy/my-app --to deploy/backend # Live traffic inspection linkerd viz tap deploy/my-app --to deploy/backend # Service edges (dependencies) linkerd viz edges deployment -n my-namespace ``` ### Template 5: Grafana Dashboard JSON ```json { "dashboard": { "title": "Service Mesh Overview", "panels": [ { "title": "Request Rate", "type": "graph", "targets": [ { "expr": "sum(rate(istio_requests_total{reporter=\"destination\"}[5m])) by (destination_service_name)", "legendFormat": "{{destination_service_name}}" } ] }, { "title": "Error Rate", "type": "gauge", "targets": [ { "expr": "sum(rate(istio_requests_total{response_code=~\"5..\"}[5m])) / sum(rate(istio_requests_total[5m])) * 100" } ], "fieldConfig": { "defaults": { "thresholds": { "steps": [ { "value": 0, "color": "green" }, { "value": 1, "color": "yellow" }, { "value": 5, "color": "red" } ] } } } }, { "title": "P99 Latency", "type": "graph", "targets": [ { "expr": "histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket{reporter=\"destination\"}[5m])) by (le, destination_service_name))", "legendFormat": "{{destination_service_name}}" } ] }, { "title": "Service Topology", "type": "nodeGraph", "targets": [ { "expr": "sum(rate(istio_requests_total{reporter=\"destination\"}[5m])) by (source_workload, destination_service_name)" } ] } ] } } ``` ### Template 6: Kiali Service Mesh Visualization ```yaml # Kiali installation apiVersion: kiali.io/v1alpha1 kind: Kiali metadata: name: kiali namespace: istio-system spec: auth: strategy: anonymous # or openid, token deployment: accessible_namespaces: - "**" external_services: prometheus: url: http://prometheus.istio-system:9090 tracing: url: http://jaeger-query.istio-system:16686 grafana: url: http://grafana.istio-system:3000 ``` ### Template 7: OpenTelemetry Integration ```yaml # OpenTelemetry Collector for mesh apiVersion: v1 kind: ConfigMap metadata: name: otel-collector-config data: config.yaml: | receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 zipkin: endpoint: 0.0.0.0:9411 processors: batch: timeout: 10s exporters: jaeger: endpoint: jaeger-collector:14250 tls: insecure: true prometheus: endpoint: 0.0.0.0:8889 service: pipelines: traces: receivers: [otlp, zipkin] processors: [batch] exporters: [jaeger] metrics: receivers: [otlp] processors: [batch] exporters: [prometheus] --- # Istio Telemetry v2 with OTel apiVersion: telemetry.istio.io/v1alpha1 kind: Telemetry metadata: name: mesh-default namespace: istio-system spec: tracing: - providers: - name: otel randomSamplingPercentage: 10 ``` ## Alerting Rules ```yaml apiVersion: monitoring.coreos.com/v1 kind: PrometheusRule metadata: name: mesh-alerts namespace: istio-system spec: groups: - name: mesh.rules rules: - alert: HighErrorRate expr: | sum(rate(istio_requests_total{response_code=~"5.."}[5m])) by (destination_service_name) / sum(rate(istio_requests_total[5m])) by (destination_service_name) > 0.05 for: 5m labels: severity: critical annotations: summary: "High error rate for {{ $labels.destination_service_name }}" - alert: HighLatency expr: | histogram_quantile(0.99, sum(rate(istio_request_duration_milliseconds_bucket[5m])) by (le, destination_service_name)) > 1000 for: 5m labels: severity: warning annotations: summary: "High P99 latency for {{ $labels.destination_service_name }}" - alert: MeshCertExpiring expr: | (certmanager_certificate_expiration_timestamp_seconds - time()) / 86400 < 7 labels: severity: warning annotations: summary: "Mesh certificate expiring in less than 7 days" ``` ## Best Practices ### Do's - **Sample appropriately** - 100% in dev, 1-10% in prod - **Use trace context** - Propagate headers consistently - **Set up alerts** - For golden signals - **Correlate metrics/traces** - Use exemplars - **Retain strategically** - Hot/cold storage tiers ### Don'ts - **Don't over-sample** - Storage costs add up - **Don't ignore cardinality** - Limit label values - **Don't skip dashboards** - Visualize dependencies - **Don't forget costs** - Monitor observability costs ## Resources - [Istio Observability](https://istio.io/latest/docs/tasks/observability/) - [Linkerd Observability](https://linkerd.io/2.14/features/dashboard/) - [OpenTelemetry](https://opentelemetry.io/) - [Kiali](https://kiali.io/)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

data-quality-frameworks

Implement data quality validation with Great Expectations, dbt

data
โญ1
# Data Quality Frameworks Production patterns for implementing data quality with Great Expectations, dbt tests, and data contracts to ensure reliable data pipelines. ## When to Use This Skill - Implementing data quality checks in pipelines - Setting up Great Expectations validation - Building comprehensive dbt test suites - Establishing data contracts between teams - Monitoring data quality metrics - Automating data validation in CI/CD ## Core Concepts ### 1. Data Quality Dimensions | Dimension | Description | Example Check | | ---------------- | ------------------------ | -------------------------------------------------- | | **Completeness** | No missing values | `expect_column_values_to_not_be_null` | | **Uniqueness** | No duplicates | `expect_column_values_to_be_unique` | | **Validity** | Values in expected range | `expect_column_values_to_be_in_set` | | **Accuracy** | Data matches reality | Cross-reference validation | | **Consistency** | No contradictions | `expect_column_pair_values_A_to_be_greater_than_B` | | **Timeliness** | Data is recent | `expect_column_max_to_be_between` | ### 2. Testing Pyramid for Data ``` /\ / \ Integration Tests (cross-table) /โ”€โ”€โ”€โ”€\ / \ Unit Tests (single column) /โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€\ / \ Schema Tests (structure) /โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€\ ``` ## Quick Start ### Great Expectations Setup ```bash # Install pip install great_expectations # Initialize project great_expectations init # Create datasource great_expectations datasource new ``` ```python # great_expectations/checkpoints/daily_validation.yml import great_expectations as gx # Create context context = gx.get_context() # Create expectation suite suite = context.add_expectation_suite("orders_suite") # Add expectations suite.add_expectation( gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id") ) suite.add_expectation( gx.expectations.ExpectColumnValuesToBeUnique(column="order_id") ) # Validate results = context.run_checkpoint(checkpoint_name="daily_orders") ``` ## Patterns ### Pattern 1: Great Expectations Suite ```python # expectations/orders_suite.py import great_expectations as gx from great_expectations.core import ExpectationSuite from great_expectations.core.expectation_configuration import ExpectationConfiguration def build_orders_suite() -> ExpectationSuite: """Build comprehensive orders expectation suite""" suite = ExpectationSuite(expectation_suite_name="orders_suite") # Schema expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_columns_to_match_set", kwargs={ "column_set": ["order_id", "customer_id", "amount", "status", "created_at"], "exact_match": False # Allow additional columns } )) # Primary key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "order_id"} )) suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_unique", kwargs={"column": "order_id"} )) # Foreign key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "customer_id"} )) # Categorical values suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_in_set", kwargs={ "column": "status", "value_set": ["pending", "processing", "shipped", "delivered", "cancelled"] } )) # Numeric ranges suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_between", kwargs={ "column": "amount", "min_value": 0, "max_value": 100000, "strict_min": True # amount > 0 } )) # Date validity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_dateutil_parseable", kwargs={"column": "created_at"} )) # Freshness - data should be recent suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_max_to_be_between", kwargs={ "column": "created_at", "min_value": {"$PARAMETER": "now - timedelta(days=1)"}, "max_value": {"$PARAMETER": "now"} } )) # Row count sanity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_row_count_to_be_between", kwargs={ "min_value": 1000, # Expect at least 1000 rows "max_value": 10000000 } )) # Statistical expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_mean_to_be_between", kwargs={ "column": "amount", "min_value": 50, "max_value": 500 } )) return suite ``` ### Pattern 2: Great Expectations Checkpoint ```yaml # great_expectations/checkpoints/orders_checkpoint.yml name: orders_checkpoint config_version: 1.0 class_name: Checkpoint run_name_template: "%Y%m%d-%H%M%S-orders-validation" validations: - batch_request: datasource_name: warehouse data_connector_name: default_inferred_data_connector_name data_asset_name: orders data_connector_query: index: -1 # Latest batch expectation_suite_name: orders_suite action_list: - name: store_validation_result action: class_name: StoreValidationResultAction - name: store_evaluation_parameters action: class_name: StoreEvaluationParametersAction - name: update_data_docs action: class_name: UpdateDataDocsAction # Slack notification on failure - name: send_slack_notification action: class_name: SlackNotificationAction slack_webhook: ${SLACK_WEBHOOK} notify_on: failure renderer: module_name: great_expectations.render.renderer.slack_renderer class_name: SlackRenderer ``` ```python # Run checkpoint import great_expectations as gx context = gx.get_context() result = context.run_checkpoint(checkpoint_name="orders_checkpoint") if not result.success: failed_expectations = [ r for r in result.run_results.values() if not r.success ] raise ValueError(f"Data quality check failed: {failed_expectations}") ``` ### Pattern 3: dbt Data Tests ```yaml # models/marts/core/_core__models.yml version: 2 models: - name: fct_orders description: Order fact table tests: # Table-level tests - dbt_utils.recency: datepart: day field: created_at interval: 1 - dbt_utils.at_least_one - dbt_utils.expression_is_true: expression: "total_amount >= 0" columns: - name: order_id description: Primary key tests: - unique - not_null - name: customer_id description: Foreign key to dim_customers tests: - not_null - relationships: to: ref('dim_customers') field: customer_id - name: order_status tests: - accepted_values: values: ["pending", "processing", "shipped", "delivered", "cancelled"] - name: total_amount tests: - not_null - dbt_utils.expression_is_true: expression: ">= 0" - name: created_at tests: - not_null - dbt_utils.expression_is_true: expression: "<= current_timestamp" - name: dim_customers columns: - name: customer_id tests: - unique - not_null - name: email tests: - unique - not_null # Custom regex test - dbt_utils.expression_is_true: expression: "email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'" ``` ### Pattern 4: Custom dbt Tests ```sql -- tests/generic/test_row_count_in_range.sql {% test row_count_in_range(model, min_count, max_count) %} with row_count as ( select count(*) as cnt from {{ model }} ) select cnt from row_count where cnt < {{ min_count }} or cnt > {{ max_count }} {% endtest %} -- Usage in schema.yml: -- tests: -- - row_count_in_range: -- min_count: 1000 -- max_count: 10000000 ``` ```sql -- tests/generic/test_sequential_values.sql {% test sequential_values(model, column_name, interval=1) %} with lagged as ( select {{ column_name }}, lag({{ column_name }}) over (order by {{ column_name }}) as prev_value from {{ model }} ) select * from lagged where {{ column_name }} - prev_value != {{ interval }} and prev_value is not null {% endtest %} ``` ```sql -- tests/singular/assert_orders_customers_match.sql -- Singular test: specific business rule with orders_customers as ( select distinct customer_id from {{ ref('fct_orders') }} ), dim_customers as ( select customer_id from {{ ref('dim_customers') }} ), orphaned_orders as ( select o.customer_id from orders_customers o left join dim_customers c using (customer_id) where c.customer_id is null ) select * from orphaned_orders -- Test passes if this returns 0 rows ``` ### Pattern 5: Data Contracts ```yaml # contracts/orders_contract.yaml apiVersion: datacontract.com/v1.0.0 kind: DataContract metadata: name: orders version: 1.0.0 owner: data-platform-team contact: data-team@company.com info: title: Orders Data Contract description: Contract for order event data from the ecommerce platform purpose: Analytics, reporting, and ML features servers: production: type: snowflake account: company.us-east-1 database: ANALYTICS schema: CORE terms: usage: Internal analytics only limitations: PII must not be exposed in downstream marts billing: Charged per query TB scanned schema: type: object properties: order_id: type: string format: uuid description: Unique order identifier required: true unique: true pii: false customer_id: type: string format: uuid description: Customer identifier required: true pii: true piiClassification: indirect total_amount: type: number minimum: 0 maximum: 100000 description: Order total in USD created_at: type: string format: date-time description: Order creation timestamp required: true status: type: string enum: [pending, processing, shipped, delivered, cancelled] description: Current order status quality: type: SodaCL specification: checks for orders: - row_count > 0 - missing_count(order_id) = 0 - duplicate_count(order_id) = 0 - invalid_count(status) = 0: valid values: [pending, processing, shipped, delivered, cancelled] - freshness(created_at) < 24h sla: availability: 99.9% freshness: 1 hour latency: 5 minutes ``` ### Pattern 6: Automated Quality Pipeline ```python # quality_pipeline.py from dataclasses import dataclass from typing import List, Dict, Any import great_expectations as gx from datetime import datetime @dataclass class QualityResult: table: str passed: bool total_expectations: int failed_expectations: int details: List[Dict[str, Any]] timestamp: datetime class DataQualityPipeline: """Orchestrate data quality checks across tables""" def __init__(self, context: gx.DataContext): self.context = context self.results: List[QualityResult] = [] def validate_table(self, table: str, suite: str) -> QualityResult: """Validate a single table against expectation suite""" checkpoint_config = { "name": f"{table}_validation", "config_version": 1.0, "class_name": "Checkpoint", "validations": [{ "batch_request": { "datasource_name": "warehouse", "data_asset_name": table, }, "expectation_suite_name": suite, }], } result = self.context.run_checkpoint(**checkpoint_config) # Parse results validation_result = list(result.run_results.values())[0] results = validation_result.results failed = [r for r in results if not r.success] return QualityResult( table=table, passed=result.success, total_expectations=len(results), failed_expectations=len(failed), details=[{ "expectation": r.expectation_config.expectation_type, "success": r.success, "observed_value": r.result.get("observed_value"), } for r in results], timestamp=datetime.now() ) def run_all(self, tables: Dict[str, str]) -> Dict[str, QualityResult]: """Run validation for all tables""" results = {} for table, suite in tables.items(): print(f"Validating {table}...") results[table] = self.validate_table(table, suite) return results def generate_report(self, results: Dict[str, QualityResult]) -> str: """Generate quality report""" report = ["# Data Quality Report", f"Generated: {datetime.now()}", ""] total_passed = sum(1 for r in results.values() if r.passed) total_tables = len(results) report.append(f"## Summary: {total_passed}/{total_tables} tables passed") report.append("") for table, result in results.items(): status = "โœ…" if result.passed else "โŒ" report.append(f"### {status} {table}") report.append(f"- Expectations: {result.total_expectations}") report.append(f"- Failed: {result.failed_expectations}") if not result.passed: report.append("- Failed checks:") for detail in result.details: if not detail["success"]: report.append(f" - {detail['expectation']}: {detail['observed_value']}") report.append("") return "\n".join(report) # Usage context = gx.get_context() pipeline = DataQualityPipeline(context) tables_to_validate = { "orders": "orders_suite", "customers": "customers_suite", "products": "products_suite", } results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results) # Fail pipeline if any table failed if not all(r.passed for r in results.values()): print(report) raise ValueError("Data quality checks failed!") ``` ## Best Practices ### Do's - **Test early** - Validate source data before transformations - **Test incrementally** - Add tests as you find issues - **Document expectations** - Clear descriptions for each test - **Alert on failures** - Integrate with monitoring - **Version contracts** - Track schema changes ### Don'ts - **Don't test everything** - Focus on critical columns - **Don't ignore warnings** - They often precede failures - **Don't skip freshness** - Stale data is bad data - **Don't hardcode thresholds** - Use dynamic baselines - **Don't test in isolation** - Test relationships too ## Resources - [Great Expectations Documentation](https://docs.greatexpectations.io/) - [dbt Testing Documentation](https://docs.getdbt.com/docs/build/tests) - [Data Contract Specification](https://datacontract.com/) - [Soda Core](https://docs.soda.io/soda-core/overview.html)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

sql-optimization-patterns

Master SQL query optimization, indexing strategies, and EXPLAIN

coding
โญ1
# SQL Optimization Patterns Transform slow database queries into lightning-fast operations through systematic optimization, proper indexing, and query plan analysis. ## When to Use This Skill - Debugging slow-running queries - Designing performant database schemas - Optimizing application response times - Reducing database load and costs - Improving scalability for growing datasets - Analyzing EXPLAIN query plans - Implementing efficient indexes - Resolving N+1 query problems ## Core Concepts ### 1. Query Execution Plans (EXPLAIN) Understanding EXPLAIN output is fundamental to optimization. **PostgreSQL EXPLAIN:** ```sql -- Basic explain EXPLAIN SELECT * FROM users WHERE email = 'user@example.com'; -- With actual execution stats EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'user@example.com'; -- Verbose output with more details EXPLAIN (ANALYZE, BUFFERS, VERBOSE) SELECT u.*, o.order_total FROM users u JOIN orders o ON u.id = o.user_id WHERE u.created_at > NOW() - INTERVAL '30 days'; ``` **Key Metrics to Watch:** - **Seq Scan**: Full table scan (usually slow for large tables) - **Index Scan**: Using index (good) - **Index Only Scan**: Using index without touching table (best) - **Nested Loop**: Join method (okay for small datasets) - **Hash Join**: Join method (good for larger datasets) - **Merge Join**: Join method (good for sorted data) - **Cost**: Estimated query cost (lower is better) - **Rows**: Estimated rows returned - **Actual Time**: Real execution time ### 2. Index Strategies Indexes are the most powerful optimization tool. **Index Types:** - **B-Tree**: Default, good for equality and range queries - **Hash**: Only for equality (=) comparisons - **GIN**: Full-text search, array queries, JSONB - **GiST**: Geometric data, full-text search - **BRIN**: Block Range INdex for very large tables with correlation ```sql -- Standard B-Tree index CREATE INDEX idx_users_email ON users(email); -- Composite index (order matters!) CREATE INDEX idx_orders_user_status ON orders(user_id, status); -- Partial index (index subset of rows) CREATE INDEX idx_active_users ON users(email) WHERE status = 'active'; -- Expression index CREATE INDEX idx_users_lower_email ON users(LOWER(email)); -- Covering index (include additional columns) CREATE INDEX idx_users_email_covering ON users(email) INCLUDE (name, created_at); -- Full-text search index CREATE INDEX idx_posts_search ON posts USING GIN(to_tsvector('english', title || ' ' || body)); -- JSONB index CREATE INDEX idx_metadata ON events USING GIN(metadata); ``` ### 3. Query Optimization Patterns **Avoid SELECT \*:** ```sql -- Bad: Fetches unnecessary columns SELECT * FROM users WHERE id = 123; -- Good: Fetch only what you need SELECT id, email, name FROM users WHERE id = 123; ``` **Use WHERE Clause Efficiently:** ```sql -- Bad: Function prevents index usage SELECT * FROM users WHERE LOWER(email) = 'user@example.com'; -- Good: Create functional index or use exact match CREATE INDEX idx_users_email_lower ON users(LOWER(email)); -- Then: SELECT * FROM users WHERE LOWER(email) = 'user@example.com'; -- Or store normalized data SELECT * FROM users WHERE email = 'user@example.com'; ``` **Optimize JOINs:** ```sql -- Bad: Cartesian product then filter SELECT u.name, o.total FROM users u, orders o WHERE u.id = o.user_id AND u.created_at > '2024-01-01'; -- Good: Filter before join SELECT u.name, o.total FROM users u JOIN orders o ON u.id = o.user_id WHERE u.created_at > '2024-01-01'; -- Better: Filter both tables SELECT u.name, o.total FROM (SELECT * FROM users WHERE created_at > '2024-01-01') u JOIN orders o ON u.id = o.user_id; ``` ## Optimization Patterns ### Pattern 1: Eliminate N+1 Queries **Problem: N+1 Query Anti-Pattern** ```python # Bad: Executes N+1 queries users = db.query("SELECT * FROM users LIMIT 10") for user in users: orders = db.query("SELECT * FROM orders WHERE user_id = ?", user.id) # Process orders ``` **Solution: Use JOINs or Batch Loading** ```sql -- Solution 1: JOIN SELECT u.id, u.name, o.id as order_id, o.total FROM users u LEFT JOIN orders o ON u.id = o.user_id WHERE u.id IN (1, 2, 3, 4, 5); -- Solution 2: Batch query SELECT * FROM orders WHERE user_id IN (1, 2, 3, 4, 5); ``` ```python # Good: Single query with JOIN or batch load # Using JOIN results = db.query(""" SELECT u.id, u.name, o.id as order_id, o.total FROM users u LEFT JOIN orders o ON u.id = o.user_id WHERE u.id IN (1, 2, 3, 4, 5) """) # Or batch load users = db.query("SELECT * FROM users LIMIT 10") user_ids = [u.id for u in users] orders = db.query( "SELECT * FROM orders WHERE user_id IN (?)", user_ids ) # Group orders by user_id orders_by_user = {} for order in orders: orders_by_user.setdefault(order.user_id, []).append(order) ``` ### Pattern 2: Optimize Pagination **Bad: OFFSET on Large Tables** ```sql -- Slow for large offsets SELECT * FROM users ORDER BY created_at DESC LIMIT 20 OFFSET 100000; -- Very slow! ``` **Good: Cursor-Based Pagination** ```sql -- Much faster: Use cursor (last seen ID) SELECT * FROM users WHERE created_at < '2024-01-15 10:30:00' -- Last cursor ORDER BY created_at DESC LIMIT 20; -- With composite sorting SELECT * FROM users WHERE (created_at, id) < ('2024-01-15 10:30:00', 12345) ORDER BY created_at DESC, id DESC LIMIT 20; -- Requires index CREATE INDEX idx_users_cursor ON users(created_at DESC, id DESC); ``` ### Pattern 3: Aggregate Efficiently **Optimize COUNT Queries:** ```sql -- Bad: Counts all rows SELECT COUNT(*) FROM orders; -- Slow on large tables -- Good: Use estimates for approximate counts SELECT reltuples::bigint AS estimate FROM pg_class WHERE relname = 'orders'; -- Good: Filter before counting SELECT COUNT(*) FROM orders WHERE created_at > NOW() - INTERVAL '7 days'; -- Better: Use index-only scan CREATE INDEX idx_orders_created ON orders(created_at); SELECT COUNT(*) FROM orders WHERE created_at > NOW() - INTERVAL '7 days'; ``` **Optimize GROUP BY:** ```sql -- Bad: Group by then filter SELECT user_id, COUNT(*) as order_count FROM orders GROUP BY user_id HAVING COUNT(*) > 10; -- Better: Filter first, then group (if possible) SELECT user_id, COUNT(*) as order_count FROM orders WHERE status = 'completed' GROUP BY user_id HAVING COUNT(*) > 10; -- Best: Use covering index CREATE INDEX idx_orders_user_status ON orders(user_id, status); ``` ### Pattern 4: Subquery Optimization **Transform Correlated Subqueries:** ```sql -- Bad: Correlated subquery (runs for each row) SELECT u.name, u.email, (SELECT COUNT(*) FROM orders o WHERE o.user_id = u.id) as order_count FROM users u; -- Good: JOIN with aggregation SELECT u.name, u.email, COUNT(o.id) as order_count FROM users u LEFT JOIN orders o ON o.user_id = u.id GROUP BY u.id, u.name, u.email; -- Better: Use window functions SELECT DISTINCT ON (u.id) u.name, u.email, COUNT(o.id) OVER (PARTITION BY u.id) as order_count FROM users u LEFT JOIN orders o ON o.user_id = u.id; ``` **Use CTEs for Clarity:** ```sql -- Using Common Table Expressions WITH recent_users AS ( SELECT id, name, email FROM users WHERE created_at > NOW() - INTERVAL '30 days' ), user_order_counts AS ( SELECT user_id, COUNT(*) as order_count FROM orders WHERE created_at > NOW() - INTERVAL '30 days' GROUP BY user_id ) SELECT ru.name, ru.email, COALESCE(uoc.order_count, 0) as orders FROM recent_users ru LEFT JOIN user_order_counts uoc ON ru.id = uoc.user_id; ``` ### Pattern 5: Batch Operations **Batch INSERT:** ```sql -- Bad: Multiple individual inserts INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com'); INSERT INTO users (name, email) VALUES ('Bob', 'bob@example.com'); INSERT INTO users (name, email) VALUES ('Carol', 'carol@example.com'); -- Good: Batch insert INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com'), ('Bob', 'bob@example.com'), ('Carol', 'carol@example.com'); -- Better: Use COPY for bulk inserts (PostgreSQL) COPY users (name, email) FROM '/tmp/users.csv' CSV HEADER; ``` **Batch UPDATE:** ```sql -- Bad: Update in loop UPDATE users SET status = 'active' WHERE id = 1; UPDATE users SET status = 'active' WHERE id = 2; -- ... repeat for many IDs -- Good: Single UPDATE with IN clause UPDATE users SET status = 'active' WHERE id IN (1, 2, 3, 4, 5, ...); -- Better: Use temporary table for large batches CREATE TEMP TABLE temp_user_updates (id INT, new_status VARCHAR); INSERT INTO temp_user_updates VALUES (1, 'active'), (2, 'active'), ...; UPDATE users u SET status = t.new_status FROM temp_user_updates t WHERE u.id = t.id; ``` ## Advanced Techniques ### Materialized Views Pre-compute expensive queries. ```sql -- Create materialized view CREATE MATERIALIZED VIEW user_order_summary AS SELECT u.id, u.name, COUNT(o.id) as total_orders, SUM(o.total) as total_spent, MAX(o.created_at) as last_order_date FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.id, u.name; -- Add index to materialized view CREATE INDEX idx_user_summary_spent ON user_order_summary(total_spent DESC); -- Refresh materialized view REFRESH MATERIALIZED VIEW user_order_summary; -- Concurrent refresh (PostgreSQL) REFRESH MATERIALIZED VIEW CONCURRENTLY user_order_summary; -- Query materialized view (very fast) SELECT * FROM user_order_summary WHERE total_spent > 1000 ORDER BY total_spent DESC; ``` ### Partitioning Split large tables for better performance. ```sql -- Range partitioning by date (PostgreSQL) CREATE TABLE orders ( id SERIAL, user_id INT, total DECIMAL, created_at TIMESTAMP ) PARTITION BY RANGE (created_at); -- Create partitions CREATE TABLE orders_2024_q1 PARTITION OF orders FOR VALUES FROM ('2024-01-01') TO ('2024-04-01'); CREATE TABLE orders_2024_q2 PARTITION OF orders FOR VALUES FROM ('2024-04-01') TO ('2024-07-01'); -- Queries automatically use appropriate partition SELECT * FROM orders WHERE created_at BETWEEN '2024-02-01' AND '2024-02-28'; -- Only scans orders_2024_q1 partition ``` ### Query Hints and Optimization ```sql -- Force index usage (MySQL) SELECT * FROM users USE INDEX (idx_users_email) WHERE email = 'user@example.com'; -- Parallel query (PostgreSQL) SET max_parallel_workers_per_gather = 4; SELECT * FROM large_table WHERE condition; -- Join hints (PostgreSQL) SET enable_nestloop = OFF; -- Force hash or merge join ``` ## Best Practices 1. **Index Selectively**: Too many indexes slow down writes 2. **Monitor Query Performance**: Use slow query logs 3. **Keep Statistics Updated**: Run ANALYZE regularly 4. **Use Appropriate Data Types**: Smaller types = better performance 5. **Normalize Thoughtfully**: Balance normalization vs performance 6. **Cache Frequently Accessed Data**: Use application-level caching 7. **Connection Pooling**: Reuse database connections 8. **Regular Maintenance**: VACUUM, ANALYZE, rebuild indexes ```sql -- Update statistics ANALYZE users; ANALYZE VERBOSE orders; -- Vacuum (PostgreSQL) VACUUM ANALYZE users; VACUUM FULL users; -- Reclaim space (locks table) -- Reindex REINDEX INDEX idx_users_email; REINDEX TABLE users; ``` ## Common Pitfalls - **Over-Indexing**: Each index slows down INSERT/UPDATE/DELETE - **Unused Indexes**: Waste space and slow writes - **Missing Indexes**: Slow queries, full table scans - **Implicit Type Conversion**: Prevents index usage - **OR Conditions**: Can't use indexes efficiently - **LIKE with Leading Wildcard**: `LIKE '%abc'` can't use index - **Function in WHERE**: Prevents index usage unless functional index exists ## Monitoring Queries ```sql -- Find slow queries (PostgreSQL) SELECT query, calls, total_time, mean_time FROM pg_stat_statements ORDER BY mean_time DESC LIMIT 10; -- Find missing indexes (PostgreSQL) SELECT schemaname, tablename, seq_scan, seq_tup_read, idx_scan, seq_tup_read / seq_scan AS avg_seq_tup_read FROM pg_stat_user_tables WHERE seq_scan > 0 ORDER BY seq_tup_read DESC LIMIT 10; -- Find unused indexes (PostgreSQL) SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch FROM pg_stat_user_indexes WHERE idx_scan = 0 ORDER BY pg_relation_size(indexrelid) DESC; ``` ## Resources - **references/postgres-optimization-guide.md**: PostgreSQL-specific optimization - **references/mysql-optimization-guide.md**: MySQL/MariaDB optimization - **references/query-plan-analysis.md**: Deep dive into EXPLAIN plans - **assets/index-strategy-checklist.md**: When and how to create indexes - **assets/query-optimization-checklist.md**: Step-by-step optimization guide - **scripts/analyze-slow-queries.sql**: Identify slow queries in your database - **scripts/index-recommendations.sql**: Generate index recommendations
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

react-modernization

Upgrade React applications to latest versions, migrate from class

coding
โญ1
# React Modernization Master React version upgrades, class to hooks migration, concurrent features adoption, and codemods for automated transformation. ## When to Use This Skill - Upgrading React applications to latest versions - Migrating class components to functional components with hooks - Adopting concurrent React features (Suspense, transitions) - Applying codemods for automated refactoring - Modernizing state management patterns - Updating to TypeScript - Improving performance with React 18+ features ## Version Upgrade Path ### React 16 โ†’ 17 โ†’ 18 **Breaking Changes by Version:** **React 17:** - Event delegation changes - No event pooling - Effect cleanup timing - JSX transform (no React import needed) **React 18:** - Automatic batching - Concurrent rendering - Strict Mode changes (double invocation) - New root API - Suspense on server ## Class to Hooks Migration ### State Management ```javascript // Before: Class component class Counter extends React.Component { constructor(props) { super(props); this.state = { count: 0, name: "", }; } increment = () => { this.setState({ count: this.state.count + 1 }); }; render() { return ( <div> <p>Count: {this.state.count}</p> <button onClick={this.increment}>Increment</button> </div> ); } } // After: Functional component with hooks function Counter() { const [count, setCount] = useState(0); const [name, setName] = useState(""); const increment = () => { setCount(count + 1); }; return ( <div> <p>Count: {count}</p> <button onClick={increment}>Increment</button> </div> ); } ``` ### Lifecycle Methods to Hooks ```javascript // Before: Lifecycle methods class DataFetcher extends React.Component { state = { data: null, loading: true }; componentDidMount() { this.fetchData(); } componentDidUpdate(prevProps) { if (prevProps.id !== this.props.id) { this.fetchData(); } } componentWillUnmount() { this.cancelRequest(); } fetchData = async () => { const data = await fetch(`/api/${this.props.id}`); this.setState({ data, loading: false }); }; cancelRequest = () => { // Cleanup }; render() { if (this.state.loading) return <div>Loading...</div>; return <div>{this.state.data}</div>; } } // After: useEffect hook function DataFetcher({ id }) { const [data, setData] = useState(null); const [loading, setLoading] = useState(true); useEffect(() => { let cancelled = false; const fetchData = async () => { try { const response = await fetch(`/api/${id}`); const result = await response.json(); if (!cancelled) { setData(result); setLoading(false); } } catch (error) { if (!cancelled) { console.error(error); } } }; fetchData(); // Cleanup function return () => { cancelled = true; }; }, [id]); // Re-run when id changes if (loading) return <div>Loading...</div>; return <div>{data}</div>; } ``` ### Context and HOCs to Hooks ```javascript // Before: Context consumer and HOC const ThemeContext = React.createContext(); class ThemedButton extends React.Component { static contextType = ThemeContext; render() { return ( <button style={{ background: this.context.theme }}> {this.props.children} </button> ); } } // After: useContext hook function ThemedButton({ children }) { const { theme } = useContext(ThemeContext); return <button style={{ background: theme }}>{children}</button>; } // Before: HOC for data fetching function withUser(Component) { return class extends React.Component { state = { user: null }; componentDidMount() { fetchUser().then((user) => this.setState({ user })); } render() { return <Component {...this.props} user={this.state.user} />; } }; } // After: Custom hook function useUser() { const [user, setUser] = useState(null); useEffect(() => { fetchUser().then(setUser); }, []); return user; } function UserProfile() { const user = useUser(); if (!user) return <div>Loading...</div>; return <div>{user.name}</div>; } ``` ## React 18 Concurrent Features ### New Root API ```javascript // Before: React 17 import ReactDOM from "react-dom"; ReactDOM.render(<App />, document.getElementById("root")); // After: React 18 import { createRoot } from "react-dom/client"; const root = createRoot(document.getElementById("root")); root.render(<App />); ``` ### Automatic Batching ```javascript // React 18: All updates are batched function handleClick() { setCount((c) => c + 1); setFlag((f) => !f); // Only one re-render (batched) } // Even in async: setTimeout(() => { setCount((c) => c + 1); setFlag((f) => !f); // Still batched in React 18! }, 1000); // Opt out if needed import { flushSync } from "react-dom"; flushSync(() => { setCount((c) => c + 1); }); // Re-render happens here setFlag((f) => !f); // Another re-render ``` ### Transitions ```javascript import { useState, useTransition } from "react"; function SearchResults() { const [query, setQuery] = useState(""); const [results, setResults] = useState([]); const [isPending, startTransition] = useTransition(); const handleChange = (e) => { // Urgent: Update input immediately setQuery(e.target.value); // Non-urgent: Update results (can be interrupted) startTransition(() => { setResults(searchResults(e.target.value)); }); }; return ( <> <input value={query} onChange={handleChange} /> {isPending && <Spinner />} <Results data={results} /> </> ); } ``` ### Suspense for Data Fetching ```javascript import { Suspense } from "react"; // Resource-based data fetching (with React 18) const resource = fetchProfileData(); function ProfilePage() { return ( <Suspense fallback={<Loading />}> <ProfileDetails /> <Suspense fallback={<Loading />}> <ProfileTimeline /> </Suspense> </Suspense> ); } function ProfileDetails() { // This will suspend if data not ready const user = resource.user.read(); return <h1>{user.name}</h1>; } function ProfileTimeline() { const posts = resource.posts.read(); return <Timeline posts={posts} />; } ``` ## Codemods for Automation ### Run React Codemods ```bash # Rename unsafe lifecycle methods npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js src/ # Update React imports (React 17+) npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/update-react-imports.js src/ # Add error boundaries npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/error-boundaries.js src/ # For TypeScript files npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js --parser=tsx src/ # Dry run to preview changes npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js --dry --print src/ # Class to Hooks (third-party) npx codemod react/hooks/convert-class-to-function src/ ``` ### Custom Codemod Example ```javascript // custom-codemod.js module.exports = function (file, api) { const j = api.jscodeshift; const root = j(file.source); // Find setState calls root .find(j.CallExpression, { callee: { type: "MemberExpression", property: { name: "setState" }, }, }) .forEach((path) => { // Transform to useState // ... transformation logic }); return root.toSource(); }; // Run: jscodeshift -t custom-codemod.js src/ ``` ## Performance Optimization ### useMemo and useCallback ```javascript function ExpensiveComponent({ items, filter }) { // Memoize expensive calculation const filteredItems = useMemo(() => { return items.filter((item) => item.category === filter); }, [items, filter]); // Memoize callback to prevent child re-renders const handleClick = useCallback((id) => { console.log("Clicked:", id); }, []); // No dependencies, never changes return <List items={filteredItems} onClick={handleClick} />; } // Child component with memo const List = React.memo(({ items, onClick }) => { return items.map((item) => ( <Item key={item.id} item={item} onClick={onClick} /> )); }); ``` ### Code Splitting ```javascript import { lazy, Suspense } from "react"; // Lazy load components const Dashboard = lazy(() => import("./Dashboard")); const Settings = lazy(() => import("./Settings")); function App() { return ( <Suspense fallback={<Loading />}> <Routes> <Route path="/dashboard" element={<Dashboard />} /> <Route path="/settings" element={<Settings />} /> </Routes> </Suspense> ); } ``` ## TypeScript Migration ```typescript // Before: JavaScript function Button({ onClick, children }) { return <button onClick={onClick}>{children}</button>; } // After: TypeScript interface ButtonProps { onClick: () => void; children: React.ReactNode; } function Button({ onClick, children }: ButtonProps) { return <button onClick={onClick}>{children}</button>; } // Generic components interface ListProps<T> { items: T[]; renderItem: (item: T) => React.ReactNode; } function List<T>({ items, renderItem }: ListProps<T>) { return <>{items.map(renderItem)}</>; } ``` ## Migration Checklist ```markdown ### Pre-Migration - [ ] Update dependencies incrementally (not all at once) - [ ] Review breaking changes in release notes - [ ] Set up testing suite - [ ] Create feature branch ### Class โ†’ Hooks Migration - [ ] Identify class components to migrate - [ ] Start with leaf components (no children) - [ ] Convert state to useState - [ ] Convert lifecycle to useEffect - [ ] Convert context to useContext - [ ] Extract custom hooks - [ ] Test thoroughly ### React 18 Upgrade - [ ] Update to React 17 first (if needed) - [ ] Update react and react-dom to 18 - [ ] Update @types/react if using TypeScript - [ ] Change to createRoot API - [ ] Test with StrictMode (double invocation) - [ ] Address concurrent rendering issues - [ ] Adopt Suspense/Transitions where beneficial ### Performance - [ ] Identify performance bottlenecks - [ ] Add React.memo where appropriate - [ ] Use useMemo/useCallback for expensive operations - [ ] Implement code splitting - [ ] Optimize re-renders ### Testing - [ ] Update test utilities (React Testing Library) - [ ] Test with React 18 features - [ ] Check for warnings in console - [ ] Performance testing ``` ## Resources - **references/breaking-changes.md**: Version-specific breaking changes - **references/codemods.md**: Codemod usage guide - **references/hooks-migration.md**: Comprehensive hooks patterns - **references/concurrent-features.md**: React 18 concurrent features - **assets/codemod-config.json**: Codemod configurations - **assets/migration-checklist.md**: Step-by-step checklist - **scripts/apply-codemods.sh**: Automated codemod script ## Best Practices 1. **Incremental Migration**: Don't migrate everything at once 2. **Test Thoroughly**: Comprehensive testing at each step 3. **Use Codemods**: Automate repetitive transformations 4. **Start Simple**: Begin with leaf components 5. **Leverage StrictMode**: Catch issues early 6. **Monitor Performance**: Measure before and after 7. **Document Changes**: Keep migration log ## Common Pitfalls - Forgetting useEffect dependencies - Over-using useMemo/useCallback - Not handling cleanup in useEffect - Mixing class and functional patterns - Ignoring StrictMode warnings - Breaking change assumptions
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

unity-ecs-patterns

Master Unity ECS (Entity Component System) with DOTS, Jobs, and

coding
โญ1
# Unity ECS Patterns Production patterns for Unity's Data-Oriented Technology Stack (DOTS) including Entity Component System, Job System, and Burst Compiler. ## When to Use This Skill - Building high-performance Unity games - Managing thousands of entities efficiently - Implementing data-oriented game systems - Optimizing CPU-bound game logic - Converting OOP game code to ECS - Using Jobs and Burst for parallelization ## Core Concepts ### 1. ECS vs OOP | Aspect | Traditional OOP | ECS/DOTS | | ----------- | ----------------- | --------------- | | Data layout | Object-oriented | Data-oriented | | Memory | Scattered | Contiguous | | Processing | Per-object | Batched | | Scaling | Poor with count | Linear scaling | | Best for | Complex behaviors | Mass simulation | ### 2. DOTS Components ``` Entity: Lightweight ID (no data) Component: Pure data (no behavior) System: Logic that processes components World: Container for entities Archetype: Unique combination of components Chunk: Memory block for same-archetype entities ``` ## Patterns ### Pattern 1: Basic ECS Setup ```csharp using Unity.Entities; using Unity.Mathematics; using Unity.Transforms; using Unity.Burst; using Unity.Collections; // Component: Pure data, no methods public struct Speed : IComponentData { public float Value; } public struct Health : IComponentData { public float Current; public float Max; } public struct Target : IComponentData { public Entity Value; } // Tag component (zero-size marker) public struct EnemyTag : IComponentData { } public struct PlayerTag : IComponentData { } // Buffer component (variable-size array) [InternalBufferCapacity(8)] public struct InventoryItem : IBufferElementData { public int ItemId; public int Quantity; } // Shared component (grouped entities) public struct TeamId : ISharedComponentData { public int Value; } ``` ### Pattern 2: Systems with ISystem (Recommended) ```csharp using Unity.Entities; using Unity.Transforms; using Unity.Mathematics; using Unity.Burst; // ISystem: Unmanaged, Burst-compatible, highest performance [BurstCompile] public partial struct MovementSystem : ISystem { [BurstCompile] public void OnCreate(ref SystemState state) { // Require components before system runs state.RequireForUpdate<Speed>(); } [BurstCompile] public void OnUpdate(ref SystemState state) { float deltaTime = SystemAPI.Time.DeltaTime; // Simple foreach - auto-generates job foreach (var (transform, speed) in SystemAPI.Query<RefRW<LocalTransform>, RefRO<Speed>>()) { transform.ValueRW.Position += new float3(0, 0, speed.ValueRO.Value * deltaTime); } } [BurstCompile] public void OnDestroy(ref SystemState state) { } } // With explicit job for more control [BurstCompile] public partial struct MovementJobSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { var job = new MoveJob { DeltaTime = SystemAPI.Time.DeltaTime }; state.Dependency = job.ScheduleParallel(state.Dependency); } } [BurstCompile] public partial struct MoveJob : IJobEntity { public float DeltaTime; void Execute(ref LocalTransform transform, in Speed speed) { transform.Position += new float3(0, 0, speed.Value * DeltaTime); } } ``` ### Pattern 3: Entity Queries ```csharp [BurstCompile] public partial struct QueryExamplesSystem : ISystem { private EntityQuery _enemyQuery; public void OnCreate(ref SystemState state) { // Build query manually for complex cases _enemyQuery = new EntityQueryBuilder(Allocator.Temp) .WithAll<EnemyTag, Health, LocalTransform>() .WithNone<Dead>() .WithOptions(EntityQueryOptions.FilterWriteGroup) .Build(ref state); } [BurstCompile] public void OnUpdate(ref SystemState state) { // SystemAPI.Query - simplest approach foreach (var (health, entity) in SystemAPI.Query<RefRW<Health>>() .WithAll<EnemyTag>() .WithEntityAccess()) { if (health.ValueRO.Current <= 0) { // Mark for destruction SystemAPI.GetSingleton<EndSimulationEntityCommandBufferSystem.Singleton>() .CreateCommandBuffer(state.WorldUnmanaged) .DestroyEntity(entity); } } // Get count int enemyCount = _enemyQuery.CalculateEntityCount(); // Get all entities var enemies = _enemyQuery.ToEntityArray(Allocator.Temp); // Get component arrays var healths = _enemyQuery.ToComponentDataArray<Health>(Allocator.Temp); } } ``` ### Pattern 4: Entity Command Buffers (Structural Changes) ```csharp // Structural changes (create/destroy/add/remove) require command buffers [BurstCompile] [UpdateInGroup(typeof(SimulationSystemGroup))] public partial struct SpawnSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { var ecbSingleton = SystemAPI.GetSingleton<BeginSimulationEntityCommandBufferSystem.Singleton>(); var ecb = ecbSingleton.CreateCommandBuffer(state.WorldUnmanaged); foreach (var (spawner, transform) in SystemAPI.Query<RefRW<Spawner>, RefRO<LocalTransform>>()) { spawner.ValueRW.Timer -= SystemAPI.Time.DeltaTime; if (spawner.ValueRO.Timer <= 0) { spawner.ValueRW.Timer = spawner.ValueRO.Interval; // Create entity (deferred until sync point) Entity newEntity = ecb.Instantiate(spawner.ValueRO.Prefab); // Set component values ecb.SetComponent(newEntity, new LocalTransform { Position = transform.ValueRO.Position, Rotation = quaternion.identity, Scale = 1f }); // Add component ecb.AddComponent(newEntity, new Speed { Value = 5f }); } } } } // Parallel ECB usage [BurstCompile] public partial struct ParallelSpawnJob : IJobEntity { public EntityCommandBuffer.ParallelWriter ECB; void Execute([EntityIndexInQuery] int index, in Spawner spawner) { Entity e = ECB.Instantiate(index, spawner.Prefab); ECB.AddComponent(index, e, new Speed { Value = 5f }); } } ``` ### Pattern 5: Aspect (Grouping Components) ```csharp using Unity.Entities; using Unity.Transforms; using Unity.Mathematics; // Aspect: Groups related components for cleaner code public readonly partial struct CharacterAspect : IAspect { public readonly Entity Entity; private readonly RefRW<LocalTransform> _transform; private readonly RefRO<Speed> _speed; private readonly RefRW<Health> _health; // Optional component [Optional] private readonly RefRO<Shield> _shield; // Buffer private readonly DynamicBuffer<InventoryItem> _inventory; public float3 Position { get => _transform.ValueRO.Position; set => _transform.ValueRW.Position = value; } public float CurrentHealth => _health.ValueRO.Current; public float MaxHealth => _health.ValueRO.Max; public float MoveSpeed => _speed.ValueRO.Value; public bool HasShield => _shield.IsValid; public float ShieldAmount => HasShield ? _shield.ValueRO.Amount : 0f; public void TakeDamage(float amount) { float remaining = amount; if (HasShield && _shield.ValueRO.Amount > 0) { // Shield absorbs damage first remaining = math.max(0, amount - _shield.ValueRO.Amount); } _health.ValueRW.Current = math.max(0, _health.ValueRO.Current - remaining); } public void Move(float3 direction, float deltaTime) { _transform.ValueRW.Position += direction * _speed.ValueRO.Value * deltaTime; } public void AddItem(int itemId, int quantity) { _inventory.Add(new InventoryItem { ItemId = itemId, Quantity = quantity }); } } // Using aspect in system [BurstCompile] public partial struct CharacterSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { float dt = SystemAPI.Time.DeltaTime; foreach (var character in SystemAPI.Query<CharacterAspect>()) { character.Move(new float3(1, 0, 0), dt); if (character.CurrentHealth < character.MaxHealth * 0.5f) { // Low health logic } } } } ``` ### Pattern 6: Singleton Components ```csharp // Singleton: Exactly one entity with this component public struct GameConfig : IComponentData { public float DifficultyMultiplier; public int MaxEnemies; public float SpawnRate; } public struct GameState : IComponentData { public int Score; public int Wave; public float TimeRemaining; } // Create singleton on world creation public partial struct GameInitSystem : ISystem { public void OnCreate(ref SystemState state) { var entity = state.EntityManager.CreateEntity(); state.EntityManager.AddComponentData(entity, new GameConfig { DifficultyMultiplier = 1.0f, MaxEnemies = 100, SpawnRate = 2.0f }); state.EntityManager.AddComponentData(entity, new GameState { Score = 0, Wave = 1, TimeRemaining = 120f }); } } // Access singleton in system [BurstCompile] public partial struct ScoreSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { // Read singleton var config = SystemAPI.GetSingleton<GameConfig>(); // Write singleton ref var gameState = ref SystemAPI.GetSingletonRW<GameState>().ValueRW; gameState.TimeRemaining -= SystemAPI.Time.DeltaTime; // Check exists if (SystemAPI.HasSingleton<GameConfig>()) { // ... } } } ``` ### Pattern 7: Baking (Converting GameObjects) ```csharp using Unity.Entities; using UnityEngine; // Authoring component (MonoBehaviour in Editor) public class EnemyAuthoring : MonoBehaviour { public float Speed = 5f; public float Health = 100f; public GameObject ProjectilePrefab; class Baker : Baker<EnemyAuthoring> { public override void Bake(EnemyAuthoring authoring) { var entity = GetEntity(TransformUsageFlags.Dynamic); AddComponent(entity, new Speed { Value = authoring.Speed }); AddComponent(entity, new Health { Current = authoring.Health, Max = authoring.Health }); AddComponent(entity, new EnemyTag()); if (authoring.ProjectilePrefab != null) { AddComponent(entity, new ProjectilePrefab { Value = GetEntity(authoring.ProjectilePrefab, TransformUsageFlags.Dynamic) }); } } } } // Complex baking with dependencies public class SpawnerAuthoring : MonoBehaviour { public GameObject[] Prefabs; public float Interval = 1f; class Baker : Baker<SpawnerAuthoring> { public override void Bake(SpawnerAuthoring authoring) { var entity = GetEntity(TransformUsageFlags.Dynamic); AddComponent(entity, new Spawner { Interval = authoring.Interval, Timer = 0f }); // Bake buffer of prefabs var buffer = AddBuffer<SpawnPrefabElement>(entity); foreach (var prefab in authoring.Prefabs) { buffer.Add(new SpawnPrefabElement { Prefab = GetEntity(prefab, TransformUsageFlags.Dynamic) }); } // Declare dependencies DependsOn(authoring.Prefabs); } } } ``` ### Pattern 8: Jobs with Native Collections ```csharp using Unity.Jobs; using Unity.Collections; using Unity.Burst; using Unity.Mathematics; [BurstCompile] public struct SpatialHashJob : IJobParallelFor { [ReadOnly] public NativeArray<float3> Positions; // Thread-safe write to hash map public NativeParallelMultiHashMap<int, int>.ParallelWriter HashMap; public float CellSize; public void Execute(int index) { float3 pos = Positions[index]; int hash = GetHash(pos); HashMap.Add(hash, index); } int GetHash(float3 pos) { int x = (int)math.floor(pos.x / CellSize); int y = (int)math.floor(pos.y / CellSize); int z = (int)math.floor(pos.z / CellSize); return x * 73856093 ^ y * 19349663 ^ z * 83492791; } } [BurstCompile] public partial struct SpatialHashSystem : ISystem { private NativeParallelMultiHashMap<int, int> _hashMap; public void OnCreate(ref SystemState state) { _hashMap = new NativeParallelMultiHashMap<int, int>(10000, Allocator.Persistent); } public void OnDestroy(ref SystemState state) { _hashMap.Dispose(); } [BurstCompile] public void OnUpdate(ref SystemState state) { var query = SystemAPI.QueryBuilder() .WithAll<LocalTransform>() .Build(); int count = query.CalculateEntityCount(); // Resize if needed if (_hashMap.Capacity < count) { _hashMap.Capacity = count * 2; } _hashMap.Clear(); // Get positions var positions = query.ToComponentDataArray<LocalTransform>(Allocator.TempJob); var posFloat3 = new NativeArray<float3>(count, Allocator.TempJob); for (int i = 0; i < count; i++) { posFloat3[i] = positions[i].Position; } // Build hash map var hashJob = new SpatialHashJob { Positions = posFloat3, HashMap = _hashMap.AsParallelWriter(), CellSize = 10f }; state.Dependency = hashJob.Schedule(count, 64, state.Dependency); // Cleanup positions.Dispose(state.Dependency); posFloat3.Dispose(state.Dependency); } } ``` ## Performance Tips ```csharp // 1. Use Burst everywhere [BurstCompile] public partial struct MySystem : ISystem { } // 2. Prefer IJobEntity over manual iteration [BurstCompile] partial struct OptimizedJob : IJobEntity { void Execute(ref LocalTransform transform) { } } // 3. Schedule parallel when possible state.Dependency = job.ScheduleParallel(state.Dependency); // 4. Use ScheduleParallel with chunk iteration [BurstCompile] partial struct ChunkJob : IJobChunk { public ComponentTypeHandle<Health> HealthHandle; public void Execute(in ArchetypeChunk chunk, int unfilteredChunkIndex, bool useEnabledMask, in v128 chunkEnabledMask) { var healths = chunk.GetNativeArray(ref HealthHandle); for (int i = 0; i < chunk.Count; i++) { // Process } } } // 5. Avoid structural changes in hot paths // Use enableable components instead of add/remove public struct Disabled : IComponentData, IEnableableComponent { } ``` ## Best Practices ### Do's - **Use ISystem over SystemBase** - Better performance - **Burst compile everything** - Massive speedup - **Batch structural changes** - Use ECB - **Profile with Profiler** - Identify bottlenecks - **Use Aspects** - Clean component grouping ### Don'ts - **Don't use managed types** - Breaks Burst - **Don't structural change in jobs** - Use ECB - **Don't over-architect** - Start simple - **Don't ignore chunk utilization** - Group similar entities - **Don't forget disposal** - Native collections leak ## Resources - [Unity DOTS Documentation](https://docs.unity3d.com/Packages/com.unity.entities@latest) - [Unity DOTS Samples](https://github.com/Unity-Technologies/EntityComponentSystemSamples) - [Burst User Guide](https://docs.unity3d.com/Packages/com.unity.burst@latest)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

helm-chart-scaffolding

Design, organize, and manage Helm charts for templating and

architecture
โญ1
# Helm Chart Scaffolding Comprehensive guidance for creating, organizing, and managing Helm charts for packaging and deploying Kubernetes applications. ## Purpose This skill provides step-by-step instructions for building production-ready Helm charts, including chart structure, templating patterns, values management, and validation strategies. ## When to Use This Skill Use this skill when you need to: - Create new Helm charts from scratch - Package Kubernetes applications for distribution - Manage multi-environment deployments with Helm - Implement templating for reusable Kubernetes manifests - Set up Helm chart repositories - Follow Helm best practices and conventions ## Helm Overview **Helm** is the package manager for Kubernetes that: - Templates Kubernetes manifests for reusability - Manages application releases and rollbacks - Handles dependencies between charts - Provides version control for deployments - Simplifies configuration management across environments ## Step-by-Step Workflow ### 1. Initialize Chart Structure **Create new chart:** ```bash helm create my-app ``` **Standard chart structure:** ``` my-app/ โ”œโ”€โ”€ Chart.yaml # Chart metadata โ”œโ”€โ”€ values.yaml # Default configuration values โ”œโ”€โ”€ charts/ # Chart dependencies โ”œโ”€โ”€ templates/ # Kubernetes manifest templates โ”‚ โ”œโ”€โ”€ NOTES.txt # Post-install notes โ”‚ โ”œโ”€โ”€ _helpers.tpl # Template helpers โ”‚ โ”œโ”€โ”€ deployment.yaml โ”‚ โ”œโ”€โ”€ service.yaml โ”‚ โ”œโ”€โ”€ ingress.yaml โ”‚ โ”œโ”€โ”€ serviceaccount.yaml โ”‚ โ”œโ”€โ”€ hpa.yaml โ”‚ โ””โ”€โ”€ tests/ โ”‚ โ””โ”€โ”€ test-connection.yaml โ””โ”€โ”€ .helmignore # Files to ignore ``` ### 2. Configure Chart.yaml **Chart metadata defines the package:** ```yaml apiVersion: v2 name: my-app description: A Helm chart for My Application type: application version: 1.0.0 # Chart version appVersion: "2.1.0" # Application version # Keywords for chart discovery keywords: - web - api - backend # Maintainer information maintainers: - name: DevOps Team email: devops@example.com url: https://github.com/example/my-app # Source code repository sources: - https://github.com/example/my-app # Homepage home: https://example.com # Chart icon icon: https://example.com/icon.png # Dependencies dependencies: - name: postgresql version: "12.0.0" repository: "https://charts.bitnami.com/bitnami" condition: postgresql.enabled - name: redis version: "17.0.0" repository: "https://charts.bitnami.com/bitnami" condition: redis.enabled ``` **Reference:** See `assets/Chart.yaml.template` for complete example ### 3. Design values.yaml Structure **Organize values hierarchically:** ```yaml # Image configuration image: repository: myapp tag: "1.0.0" pullPolicy: IfNotPresent # Number of replicas replicaCount: 3 # Service configuration service: type: ClusterIP port: 80 targetPort: 8080 # Ingress configuration ingress: enabled: false className: nginx hosts: - host: app.example.com paths: - path: / pathType: Prefix # Resources resources: requests: memory: "256Mi" cpu: "250m" limits: memory: "512Mi" cpu: "500m" # Autoscaling autoscaling: enabled: false minReplicas: 2 maxReplicas: 10 targetCPUUtilizationPercentage: 80 # Environment variables env: - name: LOG_LEVEL value: "info" # ConfigMap data configMap: data: APP_MODE: production # Dependencies postgresql: enabled: true auth: database: myapp username: myapp redis: enabled: false ``` **Reference:** See `assets/values.yaml.template` for complete structure ### 4. Create Template Files **Use Go templating with Helm functions:** **templates/deployment.yaml:** ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: {{ include "my-app.fullname" . }} labels: {{- include "my-app.labels" . | nindent 4 }} spec: {{- if not .Values.autoscaling.enabled }} replicas: {{ .Values.replicaCount }} {{- end }} selector: matchLabels: {{- include "my-app.selectorLabels" . | nindent 6 }} template: metadata: labels: {{- include "my-app.selectorLabels" . | nindent 8 }} spec: containers: - name: {{ .Chart.Name }} image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" imagePullPolicy: {{ .Values.image.pullPolicy }} ports: - name: http containerPort: {{ .Values.service.targetPort }} resources: {{- toYaml .Values.resources | nindent 12 }} env: {{- toYaml .Values.env | nindent 12 }} ``` ### 5. Create Template Helpers **templates/\_helpers.tpl:** ```yaml {{/* Expand the name of the chart. */}} {{- define "my-app.name" -}} {{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }} {{- end }} {{/* Create a default fully qualified app name. */}} {{- define "my-app.fullname" -}} {{- if .Values.fullnameOverride }} {{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }} {{- else }} {{- $name := default .Chart.Name .Values.nameOverride }} {{- if contains $name .Release.Name }} {{- .Release.Name | trunc 63 | trimSuffix "-" }} {{- else }} {{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }} {{- end }} {{- end }} {{- end }} {{/* Common labels */}} {{- define "my-app.labels" -}} helm.sh/chart: {{ include "my-app.chart" . }} {{ include "my-app.selectorLabels" . }} {{- if .Chart.AppVersion }} app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} {{- end }} app.kubernetes.io/managed-by: {{ .Release.Service }} {{- end }} {{/* Selector labels */}} {{- define "my-app.selectorLabels" -}} app.kubernetes.io/name: {{ include "my-app.name" . }} app.kubernetes.io/instance: {{ .Release.Name }} {{- end }} ``` ### 6. Manage Dependencies **Add dependencies in Chart.yaml:** ```yaml dependencies: - name: postgresql version: "12.0.0" repository: "https://charts.bitnami.com/bitnami" condition: postgresql.enabled ``` **Update dependencies:** ```bash helm dependency update helm dependency build ``` **Override dependency values:** ```yaml # values.yaml postgresql: enabled: true auth: database: myapp username: myapp password: changeme primary: persistence: enabled: true size: 10Gi ``` ### 7. Test and Validate **Validation commands:** ```bash # Lint the chart helm lint my-app/ # Dry-run installation helm install my-app ./my-app --dry-run --debug # Template rendering helm template my-app ./my-app # Template with values helm template my-app ./my-app -f values-prod.yaml # Show computed values helm show values ./my-app ``` **Validation script:** ```bash #!/bin/bash set -e echo "Linting chart..." helm lint . echo "Testing template rendering..." helm template test-release . --dry-run echo "Checking for required values..." helm template test-release . --validate echo "All validations passed!" ``` **Reference:** See `scripts/validate-chart.sh` ### 8. Package and Distribute **Package the chart:** ```bash helm package my-app/ # Creates: my-app-1.0.0.tgz ``` **Create chart repository:** ```bash # Create index helm repo index . # Upload to repository # AWS S3 example aws s3 sync . s3://my-helm-charts/ --exclude "*" --include "*.tgz" --include "index.yaml" ``` **Use the chart:** ```bash helm repo add my-repo https://charts.example.com helm repo update helm install my-app my-repo/my-app ``` ### 9. Multi-Environment Configuration **Environment-specific values files:** ``` my-app/ โ”œโ”€โ”€ values.yaml # Defaults โ”œโ”€โ”€ values-dev.yaml # Development โ”œโ”€โ”€ values-staging.yaml # Staging โ””โ”€โ”€ values-prod.yaml # Production ``` **values-prod.yaml:** ```yaml replicaCount: 5 image: tag: "2.1.0" resources: requests: memory: "512Mi" cpu: "500m" limits: memory: "1Gi" cpu: "1000m" autoscaling: enabled: true minReplicas: 3 maxReplicas: 20 ingress: enabled: true hosts: - host: app.example.com paths: - path: / pathType: Prefix postgresql: enabled: true primary: persistence: size: 100Gi ``` **Install with environment:** ```bash helm install my-app ./my-app -f values-prod.yaml --namespace production ``` ### 10. Implement Hooks and Tests **Pre-install hook:** ```yaml # templates/pre-install-job.yaml apiVersion: batch/v1 kind: Job metadata: name: {{ include "my-app.fullname" . }}-db-setup annotations: "helm.sh/hook": pre-install "helm.sh/hook-weight": "-5" "helm.sh/hook-delete-policy": hook-succeeded spec: template: spec: containers: - name: db-setup image: postgres:15 command: ["psql", "-c", "CREATE DATABASE myapp"] restartPolicy: Never ``` **Test connection:** ```yaml # templates/tests/test-connection.yaml apiVersion: v1 kind: Pod metadata: name: "{{ include "my-app.fullname" . }}-test-connection" annotations: "helm.sh/hook": test spec: containers: - name: wget image: busybox command: ['wget'] args: ['{{ include "my-app.fullname" . }}:{{ .Values.service.port }}'] restartPolicy: Never ``` **Run tests:** ```bash helm test my-app ``` ## Common Patterns ### Pattern 1: Conditional Resources ```yaml {{- if .Values.ingress.enabled }} apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: {{ include "my-app.fullname" . }} spec: # ... {{- end }} ``` ### Pattern 2: Iterating Over Lists ```yaml env: {{- range .Values.env }} - name: {{ .name }} value: {{ .value | quote }} {{- end }} ``` ### Pattern 3: Including Files ```yaml data: config.yaml: | {{- .Files.Get "config/application.yaml" | nindent 4 }} ``` ### Pattern 4: Global Values ```yaml global: imageRegistry: docker.io imagePullSecrets: - name: regcred # Use in templates: image: {{ .Values.global.imageRegistry }}/{{ .Values.image.repository }} ``` ## Best Practices 1. **Use semantic versioning** for chart and app versions 2. **Document all values** in values.yaml with comments 3. **Use template helpers** for repeated logic 4. **Validate charts** before packaging 5. **Pin dependency versions** explicitly 6. **Use conditions** for optional resources 7. **Follow naming conventions** (lowercase, hyphens) 8. **Include NOTES.txt** with usage instructions 9. **Add labels** consistently using helpers 10. **Test installations** in all environments ## Troubleshooting **Template rendering errors:** ```bash helm template my-app ./my-app --debug ``` **Dependency issues:** ```bash helm dependency update helm dependency list ``` **Installation failures:** ```bash helm install my-app ./my-app --dry-run --debug kubectl get events --sort-by='.lastTimestamp' ``` ## Reference Files - `assets/Chart.yaml.template` - Chart metadata template - `assets/values.yaml.template` - Values structure template - `scripts/validate-chart.sh` - Validation script - `references/chart-structure.md` - Detailed chart organization ## Related Skills - `k8s-manifest-generator` - For creating base Kubernetes manifests - `gitops-workflow` - For automated Helm chart deployments
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

k8s-manifest-generator

Create production-ready Kubernetes manifests for Deployments,

architecture
โญ1
# Kubernetes Manifest Generator Step-by-step guidance for creating production-ready Kubernetes manifests including Deployments, Services, ConfigMaps, Secrets, and PersistentVolumeClaims. ## Purpose This skill provides comprehensive guidance for generating well-structured, secure, and production-ready Kubernetes manifests following cloud-native best practices and Kubernetes conventions. ## When to Use This Skill Use this skill when you need to: - Create new Kubernetes Deployment manifests - Define Service resources for network connectivity - Generate ConfigMap and Secret resources for configuration management - Create PersistentVolumeClaim manifests for stateful workloads - Follow Kubernetes best practices and naming conventions - Implement resource limits, health checks, and security contexts - Design manifests for multi-environment deployments ## Step-by-Step Workflow ### 1. Gather Requirements **Understand the workload:** - Application type (stateless/stateful) - Container image and version - Environment variables and configuration needs - Storage requirements - Network exposure requirements (internal/external) - Resource requirements (CPU, memory) - Scaling requirements - Health check endpoints **Questions to ask:** - What is the application name and purpose? - What container image and tag will be used? - Does the application need persistent storage? - What ports does the application expose? - Are there any secrets or configuration files needed? - What are the CPU and memory requirements? - Does the application need to be exposed externally? ### 2. Create Deployment Manifest **Follow this structure:** ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: <app-name> namespace: <namespace> labels: app: <app-name> version: <version> spec: replicas: 3 selector: matchLabels: app: <app-name> template: metadata: labels: app: <app-name> version: <version> spec: containers: - name: <container-name> image: <image>:<tag> ports: - containerPort: <port> name: http resources: requests: memory: "256Mi" cpu: "250m" limits: memory: "512Mi" cpu: "500m" livenessProbe: httpGet: path: /health port: http initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /ready port: http initialDelaySeconds: 5 periodSeconds: 5 env: - name: ENV_VAR value: "value" envFrom: - configMapRef: name: <app-name>-config - secretRef: name: <app-name>-secret ``` **Best practices to apply:** - Always set resource requests and limits - Implement both liveness and readiness probes - Use specific image tags (never `:latest`) - Apply security context for non-root users - Use labels for organization and selection - Set appropriate replica count based on availability needs **Reference:** See `references/deployment-spec.md` for detailed deployment options ### 3. Create Service Manifest **Choose the appropriate Service type:** **ClusterIP (internal only):** ```yaml apiVersion: v1 kind: Service metadata: name: <app-name> namespace: <namespace> labels: app: <app-name> spec: type: ClusterIP selector: app: <app-name> ports: - name: http port: 80 targetPort: 8080 protocol: TCP ``` **LoadBalancer (external access):** ```yaml apiVersion: v1 kind: Service metadata: name: <app-name> namespace: <namespace> labels: app: <app-name> annotations: service.beta.kubernetes.io/aws-load-balancer-type: nlb spec: type: LoadBalancer selector: app: <app-name> ports: - name: http port: 80 targetPort: 8080 protocol: TCP ``` **Reference:** See `references/service-spec.md` for service types and networking ### 4. Create ConfigMap **For application configuration:** ```yaml apiVersion: v1 kind: ConfigMap metadata: name: <app-name>-config namespace: <namespace> data: APP_MODE: production LOG_LEVEL: info DATABASE_HOST: db.example.com # For config files app.properties: | server.port=8080 server.host=0.0.0.0 logging.level=INFO ``` **Best practices:** - Use ConfigMaps for non-sensitive data only - Organize related configuration together - Use meaningful names for keys - Consider using one ConfigMap per component - Version ConfigMaps when making changes **Reference:** See `assets/configmap-template.yaml` for examples ### 5. Create Secret **For sensitive data:** ```yaml apiVersion: v1 kind: Secret metadata: name: <app-name>-secret namespace: <namespace> type: Opaque stringData: DATABASE_PASSWORD: "changeme" API_KEY: "secret-api-key" # For certificate files tls.crt: | -----BEGIN CERTIFICATE----- ... -----END CERTIFICATE----- tls.key: | -----BEGIN PRIVATE KEY----- ... -----END PRIVATE KEY----- ``` **Security considerations:** - Never commit secrets to Git in plain text - Use Sealed Secrets, External Secrets Operator, or Vault - Rotate secrets regularly - Use RBAC to limit secret access - Consider using Secret type: `kubernetes.io/tls` for TLS secrets ### 6. Create PersistentVolumeClaim (if needed) **For stateful applications:** ```yaml apiVersion: v1 kind: PersistentVolumeClaim metadata: name: <app-name>-data namespace: <namespace> spec: accessModes: - ReadWriteOnce storageClassName: gp3 resources: requests: storage: 10Gi ``` **Mount in Deployment:** ```yaml spec: template: spec: containers: - name: app volumeMounts: - name: data mountPath: /var/lib/app volumes: - name: data persistentVolumeClaim: claimName: <app-name>-data ``` **Storage considerations:** - Choose appropriate StorageClass for performance needs - Use ReadWriteOnce for single-pod access - Use ReadWriteMany for multi-pod shared storage - Consider backup strategies - Set appropriate retention policies ### 7. Apply Security Best Practices **Add security context to Deployment:** ```yaml spec: template: spec: securityContext: runAsNonRoot: true runAsUser: 1000 fsGroup: 1000 seccompProfile: type: RuntimeDefault containers: - name: app securityContext: allowPrivilegeEscalation: false readOnlyRootFilesystem: true capabilities: drop: - ALL ``` **Security checklist:** - [ ] Run as non-root user - [ ] Drop all capabilities - [ ] Use read-only root filesystem - [ ] Disable privilege escalation - [ ] Set seccomp profile - [ ] Use Pod Security Standards ### 8. Add Labels and Annotations **Standard labels (recommended):** ```yaml metadata: labels: app.kubernetes.io/name: <app-name> app.kubernetes.io/instance: <instance-name> app.kubernetes.io/version: "1.0.0" app.kubernetes.io/component: backend app.kubernetes.io/part-of: <system-name> app.kubernetes.io/managed-by: kubectl ``` **Useful annotations:** ```yaml metadata: annotations: description: "Application description" contact: "team@example.com" prometheus.io/scrape: "true" prometheus.io/port: "9090" prometheus.io/path: "/metrics" ``` ### 9. Organize Multi-Resource Manifests **File organization options:** **Option 1: Single file with `---` separator** ```yaml # app-name.yaml --- apiVersion: v1 kind: ConfigMap ... --- apiVersion: v1 kind: Secret ... --- apiVersion: apps/v1 kind: Deployment ... --- apiVersion: v1 kind: Service ... ``` **Option 2: Separate files** ``` manifests/ โ”œโ”€โ”€ configmap.yaml โ”œโ”€โ”€ secret.yaml โ”œโ”€โ”€ deployment.yaml โ”œโ”€โ”€ service.yaml โ””โ”€โ”€ pvc.yaml ``` **Option 3: Kustomize structure** ``` base/ โ”œโ”€โ”€ kustomization.yaml โ”œโ”€โ”€ deployment.yaml โ”œโ”€โ”€ service.yaml โ””โ”€โ”€ configmap.yaml overlays/ โ”œโ”€โ”€ dev/ โ”‚ โ””โ”€โ”€ kustomization.yaml โ””โ”€โ”€ prod/ โ””โ”€โ”€ kustomization.yaml ``` ### 10. Validate and Test **Validation steps:** ```bash # Dry-run validation kubectl apply -f manifest.yaml --dry-run=client # Server-side validation kubectl apply -f manifest.yaml --dry-run=server # Validate with kubeval kubeval manifest.yaml # Validate with kube-score kube-score score manifest.yaml # Check with kube-linter kube-linter lint manifest.yaml ``` **Testing checklist:** - [ ] Manifest passes dry-run validation - [ ] All required fields are present - [ ] Resource limits are reasonable - [ ] Health checks are configured - [ ] Security context is set - [ ] Labels follow conventions - [ ] Namespace exists or is created ## Common Patterns ### Pattern 1: Simple Stateless Web Application **Use case:** Standard web API or microservice **Components needed:** - Deployment (3 replicas for HA) - ClusterIP Service - ConfigMap for configuration - Secret for API keys - HorizontalPodAutoscaler (optional) **Reference:** See `assets/deployment-template.yaml` ### Pattern 2: Stateful Database Application **Use case:** Database or persistent storage application **Components needed:** - StatefulSet (not Deployment) - Headless Service - PersistentVolumeClaim template - ConfigMap for DB configuration - Secret for credentials ### Pattern 3: Background Job or Cron **Use case:** Scheduled tasks or batch processing **Components needed:** - CronJob or Job - ConfigMap for job parameters - Secret for credentials - ServiceAccount with RBAC ### Pattern 4: Multi-Container Pod **Use case:** Application with sidecar containers **Components needed:** - Deployment with multiple containers - Shared volumes between containers - Init containers for setup - Service (if needed) ## Templates The following templates are available in the `assets/` directory: - `deployment-template.yaml` - Standard deployment with best practices - `service-template.yaml` - Service configurations (ClusterIP, LoadBalancer, NodePort) - `configmap-template.yaml` - ConfigMap examples with different data types - `secret-template.yaml` - Secret examples (to be generated, not committed) - `pvc-template.yaml` - PersistentVolumeClaim templates ## Reference Documentation - `references/deployment-spec.md` - Detailed Deployment specification - `references/service-spec.md` - Service types and networking details ## Best Practices Summary 1. **Always set resource requests and limits** - Prevents resource starvation 2. **Implement health checks** - Ensures Kubernetes can manage your application 3. **Use specific image tags** - Avoid unpredictable deployments 4. **Apply security contexts** - Run as non-root, drop capabilities 5. **Use ConfigMaps and Secrets** - Separate config from code 6. **Label everything** - Enables filtering and organization 7. **Follow naming conventions** - Use standard Kubernetes labels 8. **Validate before applying** - Use dry-run and validation tools 9. **Version your manifests** - Keep in Git with version control 10. **Document with annotations** - Add context for other developers ## Troubleshooting **Pods not starting:** - Check image pull errors: `kubectl describe pod <pod-name>` - Verify resource availability: `kubectl get nodes` - Check events: `kubectl get events --sort-by='.lastTimestamp'` **Service not accessible:** - Verify selector matches pod labels: `kubectl get endpoints <service-name>` - Check service type and port configuration - Test from within cluster: `kubectl run debug --rm -it --image=busybox -- sh` **ConfigMap/Secret not loading:** - Verify names match in Deployment - Check namespace - Ensure resources exist: `kubectl get configmap,secret` ## Next Steps After creating manifests: 1. Store in Git repository 2. Set up CI/CD pipeline for deployment 3. Consider using Helm or Kustomize for templating 4. Implement GitOps with ArgoCD or Flux 5. Add monitoring and observability ## Related Skills - `helm-chart-scaffolding` - For templating and packaging - `gitops-workflow` - For automated deployments - `k8s-security-policies` - For advanced security configurations
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

embedding-strategies

Select and optimize embedding models for semantic search and RAG

coding
โญ1
# Embedding Strategies Guide to selecting and optimizing embedding models for vector search applications. ## When to Use This Skill - Choosing embedding models for RAG - Optimizing chunking strategies - Fine-tuning embeddings for domains - Comparing embedding model performance - Reducing embedding dimensions - Handling multilingual content ## Core Concepts ### 1. Embedding Model Comparison (2026) | Model | Dimensions | Max Tokens | Best For | | -------------------------- | ---------- | ---------- | ----------------------------------- | | **voyage-3-large** | 1024 | 32000 | Claude apps (Anthropic recommended) | | **voyage-3** | 1024 | 32000 | Claude apps, cost-effective | | **voyage-code-3** | 1024 | 32000 | Code search | | **voyage-finance-2** | 1024 | 32000 | Financial documents | | **voyage-law-2** | 1024 | 32000 | Legal documents | | **text-embedding-3-large** | 3072 | 8191 | OpenAI apps, high accuracy | | **text-embedding-3-small** | 1536 | 8191 | OpenAI apps, cost-effective | | **bge-large-en-v1.5** | 1024 | 512 | Open source, local deployment | | **all-MiniLM-L6-v2** | 384 | 256 | Fast, lightweight | | **multilingual-e5-large** | 1024 | 512 | Multi-language | ### 2. Embedding Pipeline ``` Document โ†’ Chunking โ†’ Preprocessing โ†’ Embedding Model โ†’ Vector โ†“ [Overlap, Size] [Clean, Normalize] [API/Local] ``` ## Templates ### Template 1: Voyage AI Embeddings (Recommended for Claude) ```python from langchain_voyageai import VoyageAIEmbeddings from typing import List import os # Initialize Voyage AI embeddings (recommended by Anthropic for Claude) embeddings = VoyageAIEmbeddings( model="voyage-3-large", voyage_api_key=os.environ.get("VOYAGE_API_KEY") ) def get_embeddings(texts: List[str]) -> List[List[float]]: """Get embeddings from Voyage AI.""" return embeddings.embed_documents(texts) def get_query_embedding(query: str) -> List[float]: """Get single query embedding.""" return embeddings.embed_query(query) # Specialized models for domains code_embeddings = VoyageAIEmbeddings(model="voyage-code-3") finance_embeddings = VoyageAIEmbeddings(model="voyage-finance-2") legal_embeddings = VoyageAIEmbeddings(model="voyage-law-2") ``` ### Template 2: OpenAI Embeddings ```python from openai import OpenAI from typing import List import numpy as np client = OpenAI() def get_embeddings( texts: List[str], model: str = "text-embedding-3-small", dimensions: int = None ) -> List[List[float]]: """Get embeddings from OpenAI with optional dimension reduction.""" # Handle batching for large lists batch_size = 100 all_embeddings = [] for i in range(0, len(texts), batch_size): batch = texts[i:i + batch_size] kwargs = {"input": batch, "model": model} if dimensions: # Matryoshka dimensionality reduction kwargs["dimensions"] = dimensions response = client.embeddings.create(**kwargs) embeddings = [item.embedding for item in response.data] all_embeddings.extend(embeddings) return all_embeddings def get_embedding(text: str, **kwargs) -> List[float]: """Get single embedding.""" return get_embeddings([text], **kwargs)[0] # Dimension reduction with Matryoshka embeddings def get_reduced_embedding(text: str, dimensions: int = 512) -> List[float]: """Get embedding with reduced dimensions (Matryoshka).""" return get_embedding( text, model="text-embedding-3-small", dimensions=dimensions ) ``` ### Template 3: Local Embeddings with Sentence Transformers ```python from sentence_transformers import SentenceTransformer from typing import List, Optional import numpy as np class LocalEmbedder: """Local embedding with sentence-transformers.""" def __init__( self, model_name: str = "BAAI/bge-large-en-v1.5", device: str = "cuda" ): self.model = SentenceTransformer(model_name, device=device) self.model_name = model_name def embed( self, texts: List[str], normalize: bool = True, show_progress: bool = False ) -> np.ndarray: """Embed texts with optional normalization.""" embeddings = self.model.encode( texts, normalize_embeddings=normalize, show_progress_bar=show_progress, convert_to_numpy=True ) return embeddings def embed_query(self, query: str) -> np.ndarray: """Embed a query with appropriate prefix for retrieval models.""" # BGE and similar models benefit from query prefix if "bge" in self.model_name.lower(): query = f"Represent this sentence for searching relevant passages: {query}" return self.embed([query])[0] def embed_documents(self, documents: List[str]) -> np.ndarray: """Embed documents for indexing.""" return self.embed(documents) # E5 model with instructions class E5Embedder: def __init__(self, model_name: str = "intfloat/multilingual-e5-large"): self.model = SentenceTransformer(model_name) def embed_query(self, query: str) -> np.ndarray: """E5 requires 'query:' prefix for queries.""" return self.model.encode(f"query: {query}") def embed_document(self, document: str) -> np.ndarray: """E5 requires 'passage:' prefix for documents.""" return self.model.encode(f"passage: {document}") ``` ### Template 4: Chunking Strategies ```python from typing import List, Tuple import re def chunk_by_tokens( text: str, chunk_size: int = 512, chunk_overlap: int = 50, tokenizer=None ) -> List[str]: """Chunk text by token count.""" import tiktoken tokenizer = tokenizer or tiktoken.get_encoding("cl100k_base") tokens = tokenizer.encode(text) chunks = [] start = 0 while start < len(tokens): end = start + chunk_size chunk_tokens = tokens[start:end] chunk_text = tokenizer.decode(chunk_tokens) chunks.append(chunk_text) start = end - chunk_overlap return chunks def chunk_by_sentences( text: str, max_chunk_size: int = 1000, min_chunk_size: int = 100 ) -> List[str]: """Chunk text by sentences, respecting size limits.""" import nltk sentences = nltk.sent_tokenize(text) chunks = [] current_chunk = [] current_size = 0 for sentence in sentences: sentence_size = len(sentence) if current_size + sentence_size > max_chunk_size and current_chunk: chunks.append(" ".join(current_chunk)) current_chunk = [] current_size = 0 current_chunk.append(sentence) current_size += sentence_size if current_chunk: chunks.append(" ".join(current_chunk)) return chunks def chunk_by_semantic_sections( text: str, headers_pattern: str = r'^#{1,3}\s+.+$' ) -> List[Tuple[str, str]]: """Chunk markdown by headers, preserving hierarchy.""" lines = text.split('\n') chunks = [] current_header = "" current_content = [] for line in lines: if re.match(headers_pattern, line, re.MULTILINE): if current_content: chunks.append((current_header, '\n'.join(current_content))) current_header = line current_content = [] else: current_content.append(line) if current_content: chunks.append((current_header, '\n'.join(current_content))) return chunks def recursive_character_splitter( text: str, chunk_size: int = 1000, chunk_overlap: int = 200, separators: List[str] = None ) -> List[str]: """LangChain-style recursive splitter.""" separators = separators or ["\n\n", "\n", ". ", " ", ""] def split_text(text: str, separators: List[str]) -> List[str]: if not text: return [] separator = separators[0] remaining_separators = separators[1:] if separator == "": # Character-level split return [text[i:i+chunk_size] for i in range(0, len(text), chunk_size - chunk_overlap)] splits = text.split(separator) chunks = [] current_chunk = [] current_length = 0 for split in splits: split_length = len(split) + len(separator) if current_length + split_length > chunk_size and current_chunk: chunk_text = separator.join(current_chunk) # Recursively split if still too large if len(chunk_text) > chunk_size and remaining_separators: chunks.extend(split_text(chunk_text, remaining_separators)) else: chunks.append(chunk_text) # Start new chunk with overlap overlap_splits = [] overlap_length = 0 for s in reversed(current_chunk): if overlap_length + len(s) <= chunk_overlap: overlap_splits.insert(0, s) overlap_length += len(s) else: break current_chunk = overlap_splits current_length = overlap_length current_chunk.append(split) current_length += split_length if current_chunk: chunks.append(separator.join(current_chunk)) return chunks return split_text(text, separators) ``` ### Template 5: Domain-Specific Embedding Pipeline ```python import re from typing import List, Optional from dataclasses import dataclass @dataclass class EmbeddedDocument: id: str document_id: str chunk_index: int text: str embedding: List[float] metadata: dict class DomainEmbeddingPipeline: """Pipeline for domain-specific embeddings.""" def __init__( self, embedding_model: str = "voyage-3-large", chunk_size: int = 512, chunk_overlap: int = 50, preprocessing_fn=None ): self.embeddings = VoyageAIEmbeddings(model=embedding_model) self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap self.preprocess = preprocessing_fn or self._default_preprocess def _default_preprocess(self, text: str) -> str: """Default preprocessing.""" # Remove excessive whitespace text = re.sub(r'\s+', ' ', text) # Remove special characters (customize for your domain) text = re.sub(r'[^\w\s.,!?-]', '', text) return text.strip() async def process_documents( self, documents: List[dict], id_field: str = "id", content_field: str = "content", metadata_fields: Optional[List[str]] = None ) -> List[EmbeddedDocument]: """Process documents for vector storage.""" processed = [] for doc in documents: content = doc[content_field] doc_id = doc[id_field] # Preprocess cleaned = self.preprocess(content) # Chunk chunks = chunk_by_tokens( cleaned, self.chunk_size, self.chunk_overlap ) # Create embeddings embeddings = await self.embeddings.aembed_documents(chunks) # Create records for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): metadata = {"document_id": doc_id, "chunk_index": i} # Add specified metadata fields if metadata_fields: for field in metadata_fields: if field in doc: metadata[field] = doc[field] processed.append(EmbeddedDocument( id=f"{doc_id}_chunk_{i}", document_id=doc_id, chunk_index=i, text=chunk, embedding=embedding, metadata=metadata )) return processed # Code-specific pipeline class CodeEmbeddingPipeline: """Specialized pipeline for code embeddings.""" def __init__(self): # Use Voyage's code-specific model self.embeddings = VoyageAIEmbeddings(model="voyage-code-3") def chunk_code(self, code: str, language: str) -> List[dict]: """Chunk code by functions/classes using tree-sitter.""" try: import tree_sitter_languages parser = tree_sitter_languages.get_parser(language) tree = parser.parse(bytes(code, "utf8")) chunks = [] # Extract function and class definitions self._extract_nodes(tree.root_node, code, chunks) return chunks except ImportError: # Fallback to simple chunking return [{"text": code, "type": "module"}] def _extract_nodes(self, node, source_code: str, chunks: list): """Recursively extract function/class definitions.""" if node.type in ['function_definition', 'class_definition', 'method_definition']: text = source_code[node.start_byte:node.end_byte] chunks.append({ "text": text, "type": node.type, "name": self._get_name(node), "start_line": node.start_point[0], "end_line": node.end_point[0] }) for child in node.children: self._extract_nodes(child, source_code, chunks) def _get_name(self, node) -> str: """Extract name from function/class node.""" for child in node.children: if child.type == 'identifier' or child.type == 'name': return child.text.decode('utf8') return "unknown" async def embed_with_context( self, chunk: str, context: str = "" ) -> List[float]: """Embed code with surrounding context.""" if context: combined = f"Context: {context}\n\nCode:\n{chunk}" else: combined = chunk return await self.embeddings.aembed_query(combined) ``` ### Template 6: Embedding Quality Evaluation ```python import numpy as np from typing import List, Dict def evaluate_retrieval_quality( queries: List[str], relevant_docs: List[List[str]], # List of relevant doc IDs per query retrieved_docs: List[List[str]], # List of retrieved doc IDs per query k: int = 10 ) -> Dict[str, float]: """Evaluate embedding quality for retrieval.""" def precision_at_k(relevant: set, retrieved: List[str], k: int) -> float: retrieved_k = retrieved[:k] relevant_retrieved = len(set(retrieved_k) & relevant) return relevant_retrieved / k if k > 0 else 0 def recall_at_k(relevant: set, retrieved: List[str], k: int) -> float: retrieved_k = retrieved[:k] relevant_retrieved = len(set(retrieved_k) & relevant) return relevant_retrieved / len(relevant) if relevant else 0 def mrr(relevant: set, retrieved: List[str]) -> float: for i, doc in enumerate(retrieved): if doc in relevant: return 1 / (i + 1) return 0 def ndcg_at_k(relevant: set, retrieved: List[str], k: int) -> float: dcg = sum( 1 / np.log2(i + 2) if doc in relevant else 0 for i, doc in enumerate(retrieved[:k]) ) ideal_dcg = sum(1 / np.log2(i + 2) for i in range(min(len(relevant), k))) return dcg / ideal_dcg if ideal_dcg > 0 else 0 metrics = { f"precision@{k}": [], f"recall@{k}": [], "mrr": [], f"ndcg@{k}": [] } for relevant, retrieved in zip(relevant_docs, retrieved_docs): relevant_set = set(relevant) metrics[f"precision@{k}"].append(precision_at_k(relevant_set, retrieved, k)) metrics[f"recall@{k}"].append(recall_at_k(relevant_set, retrieved, k)) metrics["mrr"].append(mrr(relevant_set, retrieved)) metrics[f"ndcg@{k}"].append(ndcg_at_k(relevant_set, retrieved, k)) return {name: np.mean(values) for name, values in metrics.items()} def compute_embedding_similarity( embeddings1: np.ndarray, embeddings2: np.ndarray, metric: str = "cosine" ) -> np.ndarray: """Compute similarity matrix between embedding sets.""" if metric == "cosine": # Normalize and compute dot product norm1 = embeddings1 / np.linalg.norm(embeddings1, axis=1, keepdims=True) norm2 = embeddings2 / np.linalg.norm(embeddings2, axis=1, keepdims=True) return norm1 @ norm2.T elif metric == "euclidean": from scipy.spatial.distance import cdist return -cdist(embeddings1, embeddings2, metric='euclidean') elif metric == "dot": return embeddings1 @ embeddings2.T else: raise ValueError(f"Unknown metric: {metric}") def compare_embedding_models( texts: List[str], models: Dict[str, callable], queries: List[str], relevant_indices: List[List[int]], k: int = 5 ) -> Dict[str, Dict[str, float]]: """Compare multiple embedding models on retrieval quality.""" results = {} for model_name, embed_fn in models.items(): # Embed all texts doc_embeddings = np.array(embed_fn(texts)) retrieved_per_query = [] for query in queries: query_embedding = np.array(embed_fn([query])[0]) # Compute similarities similarities = compute_embedding_similarity( query_embedding.reshape(1, -1), doc_embeddings, metric="cosine" )[0] # Get top-k indices top_k_indices = np.argsort(similarities)[::-1][:k] retrieved_per_query.append([str(i) for i in top_k_indices]) # Convert relevant indices to string IDs relevant_docs = [[str(i) for i in indices] for indices in relevant_indices] results[model_name] = evaluate_retrieval_quality( queries, relevant_docs, retrieved_per_query, k ) return results ``` ## Best Practices ### Do's - **Match model to use case**: Code vs prose vs multilingual - **Chunk thoughtfully**: Preserve semantic boundaries - **Normalize embeddings**: For cosine similarity search - **Batch requests**: More efficient than one-by-one - **Cache embeddings**: Avoid recomputing for static content - **Use Voyage AI for Claude apps**: Recommended by Anthropic ### Don'ts - **Don't ignore token limits**: Truncation loses information - **Don't mix embedding models**: Incompatible vector spaces - **Don't skip preprocessing**: Garbage in, garbage out - **Don't over-chunk**: Lose important context - **Don't forget metadata**: Essential for filtering and debugging ## Resources - [Voyage AI Documentation](https://docs.voyageai.com/) - [OpenAI Embeddings Guide](https://platform.openai.com/docs/guides/embeddings) - [Sentence Transformers](https://www.sbert.net/) - [MTEB Benchmar
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

langchain-architecture

Design LLM applications using LangChain 1.x and LangGraph for

coding
โญ1
# LangChain & LangGraph Architecture Master modern LangChain 1.x and LangGraph for building sophisticated LLM applications with agents, state management, memory, and tool integration. ## When to Use This Skill - Building autonomous AI agents with tool access - Implementing complex multi-step LLM workflows - Managing conversation memory and state - Integrating LLMs with external data sources and APIs - Creating modular, reusable LLM application components - Implementing document processing pipelines - Building production-grade LLM applications ## Package Structure (LangChain 1.x) ``` langchain (1.2.x) # High-level orchestration langchain-core (1.2.x) # Core abstractions (messages, prompts, tools) langchain-community # Third-party integrations langgraph # Agent orchestration and state management langchain-openai # OpenAI integrations langchain-anthropic # Anthropic/Claude integrations langchain-voyageai # Voyage AI embeddings langchain-pinecone # Pinecone vector store ``` ## Core Concepts ### 1. LangGraph Agents LangGraph is the standard for building agents in 2026. It provides: **Key Features:** - **StateGraph**: Explicit state management with typed state - **Durable Execution**: Agents persist through failures - **Human-in-the-Loop**: Inspect and modify state at any point - **Memory**: Short-term and long-term memory across sessions - **Checkpointing**: Save and resume agent state **Agent Patterns:** - **ReAct**: Reasoning + Acting with `create_react_agent` - **Plan-and-Execute**: Separate planning and execution nodes - **Multi-Agent**: Supervisor routing between specialized agents - **Tool-Calling**: Structured tool invocation with Pydantic schemas ### 2. State Management LangGraph uses TypedDict for explicit state: ```python from typing import Annotated, TypedDict from langgraph.graph import MessagesState # Simple message-based state class AgentState(MessagesState): """Extends MessagesState with custom fields.""" context: Annotated[list, "retrieved documents"] # Custom state for complex agents class CustomState(TypedDict): messages: Annotated[list, "conversation history"] context: Annotated[dict, "retrieved context"] current_step: str results: list ``` ### 3. Memory Systems Modern memory implementations: - **ConversationBufferMemory**: Stores all messages (short conversations) - **ConversationSummaryMemory**: Summarizes older messages (long conversations) - **ConversationTokenBufferMemory**: Token-based windowing - **VectorStoreRetrieverMemory**: Semantic similarity retrieval - **LangGraph Checkpointers**: Persistent state across sessions ### 4. Document Processing Loading, transforming, and storing documents: **Components:** - **Document Loaders**: Load from various sources - **Text Splitters**: Chunk documents intelligently - **Vector Stores**: Store and retrieve embeddings - **Retrievers**: Fetch relevant documents ### 5. Callbacks & Tracing LangSmith is the standard for observability: - Request/response logging - Token usage tracking - Latency monitoring - Error tracking - Trace visualization ## Quick Start ### Modern ReAct Agent with LangGraph ```python from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool import ast import operator # Initialize LLM (Claude Sonnet 4.6 recommended) llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0) # Define tools with Pydantic schemas @tool def search_database(query: str) -> str: """Search internal database for information.""" # Your database search logic return f"Results for: {query}" @tool def calculate(expression: str) -> str: """Safely evaluate a mathematical expression. Supports: +, -, *, /, **, %, parentheses Example: '(2 + 3) * 4' returns '20' """ # Safe math evaluation using ast allowed_operators = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod, ast.USub: operator.neg, } def _eval(node): if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left = _eval(node.left) right = _eval(node.right) return allowed_operators[type(node.op)](left, right) elif isinstance(node, ast.UnaryOp): operand = _eval(node.operand) return allowed_operators[type(node.op)](operand) else: raise ValueError(f"Unsupported operation: {type(node)}") try: tree = ast.parse(expression, mode='eval') return str(_eval(tree.body)) except Exception as e: return f"Error: {e}" tools = [search_database, calculate] # Create checkpointer for memory persistence checkpointer = MemorySaver() # Create ReAct agent agent = create_react_agent( llm, tools, checkpointer=checkpointer ) # Run agent with thread ID for memory config = {"configurable": {"thread_id": "user-123"}} result = await agent.ainvoke( {"messages": [("user", "Search for Python tutorials and calculate 25 * 4")]}, config=config ) ``` ## Architecture Patterns ### Pattern 1: RAG with LangGraph ```python from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from typing import TypedDict, Annotated class RAGState(TypedDict): question: str context: Annotated[list[Document], "retrieved documents"] answer: str # Initialize components llm = ChatAnthropic(model="claude-sonnet-4-6") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # Define nodes async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs} async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so. Context: {context} Question: {question} Answer:""" ) context_text = "\n\n".join(doc.page_content for doc in state["context"]) response = await llm.ainvoke( prompt.format(context=context_text, question=state["question"]) ) return {"answer": response.content} # Build graph builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) rag_chain = builder.compile() # Use the chain result = await rag_chain.ainvoke({"question": "What is the main topic?"}) ``` ### Pattern 2: Custom Agent with Structured Tools ```python from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): """Input for database search.""" query: str = Field(description="Search query") filters: dict = Field(default={}, description="Optional filters") class EmailInput(BaseModel): """Input for sending email.""" recipient: str = Field(description="Email recipient") subject: str = Field(description="Email subject") content: str = Field(description="Email body") async def search_database(query: str, filters: dict = {}) -> str: """Search internal database for information.""" # Your database search logic return f"Results for '{query}' with filters {filters}" async def send_email(recipient: str, subject: str, content: str) -> str: """Send an email to specified recipient.""" # Email sending logic return f"Email sent to {recipient}" tools = [ StructuredTool.from_function( coroutine=search_database, name="search_database", description="Search internal database", args_schema=SearchInput ), StructuredTool.from_function( coroutine=send_email, name="send_email", description="Send an email", args_schema=EmailInput ) ] agent = create_react_agent(llm, tools) ``` ### Pattern 3: Multi-Step Workflow with StateGraph ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict, Literal class WorkflowState(TypedDict): text: str entities: list analysis: str summary: str current_step: str async def extract_entities(state: WorkflowState) -> WorkflowState: """Extract key entities from text.""" prompt = f"Extract key entities from: {state['text']}\n\nReturn as JSON list." response = await llm.ainvoke(prompt) return {"entities": response.content, "current_step": "analyze"} async def analyze_entities(state: WorkflowState) -> WorkflowState: """Analyze extracted entities.""" prompt = f"Analyze these entities: {state['entities']}\n\nProvide insights." response = await llm.ainvoke(prompt) return {"analysis": response.content, "current_step": "summarize"} async def generate_summary(state: WorkflowState) -> WorkflowState: """Generate final summary.""" prompt = f"""Summarize: Entities: {state['entities']} Analysis: {state['analysis']} Provide a concise summary.""" response = await llm.ainvoke(prompt) return {"summary": response.content, "current_step": "complete"} def route_step(state: WorkflowState) -> Literal["analyze", "summarize", "end"]: """Route to next step based on current state.""" step = state.get("current_step", "extract") if step == "analyze": return "analyze" elif step == "summarize": return "summarize" return "end" # Build workflow builder = StateGraph(WorkflowState) builder.add_node("extract", extract_entities) builder.add_node("analyze", analyze_entities) builder.add_node("summarize", generate_summary) builder.add_edge(START, "extract") builder.add_conditional_edges("extract", route_step, { "analyze": "analyze", "summarize": "summarize", "end": END }) builder.add_conditional_edges("analyze", route_step, { "summarize": "summarize", "end": END }) builder.add_edge("summarize", END) workflow = builder.compile() ``` ### Pattern 4: Multi-Agent Orchestration ```python from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import create_react_agent from langchain_core.messages import HumanMessage from typing import Literal class MultiAgentState(TypedDict): messages: list next_agent: str # Create specialized agents researcher = create_react_agent(llm, research_tools) writer = create_react_agent(llm, writing_tools) reviewer = create_react_agent(llm, review_tools) async def supervisor(state: MultiAgentState) -> MultiAgentState: """Route to appropriate agent based on task.""" prompt = f"""Based on the conversation, which agent should handle this? Options: - researcher: For finding information - writer: For creating content - reviewer: For reviewing and editing - FINISH: Task is complete Messages: {state['messages']} Respond with just the agent name.""" response = await llm.ainvoke(prompt) return {"next_agent": response.content.strip().lower()} def route_to_agent(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]: """Route based on supervisor decision.""" next_agent = state.get("next_agent", "").lower() if next_agent == "finish": return "end" return next_agent if next_agent in ["researcher", "writer", "reviewer"] else "end" # Build multi-agent graph builder = StateGraph(MultiAgentState) builder.add_node("supervisor", supervisor) builder.add_node("researcher", researcher) builder.add_node("writer", writer) builder.add_node("reviewer", reviewer) builder.add_edge(START, "supervisor") builder.add_conditional_edges("supervisor", route_to_agent, { "researcher": "researcher", "writer": "writer", "reviewer": "reviewer", "end": END }) # Each agent returns to supervisor for agent in ["researcher", "writer", "reviewer"]: builder.add_edge(agent, "supervisor") multi_agent = builder.compile() ``` ## Memory Management ### Token-Based Memory with LangGraph ```python from langgraph.checkpoint.memory import MemorySaver from langgraph.prebuilt import create_react_agent # In-memory checkpointer (development) checkpointer = MemorySaver() # Create agent with persistent memory agent = create_react_agent(llm, tools, checkpointer=checkpointer) # Each thread_id maintains separate conversation config = {"configurable": {"thread_id": "session-abc123"}} # Messages persist across invocations with same thread_id result1 = await agent.ainvoke({"messages": [("user", "My name is Alice")]}, config) result2 = await agent.ainvoke({"messages": [("user", "What's my name?")]}, config) # Agent remembers: "Your name is Alice" ``` ### Production Memory with PostgreSQL ```python from langgraph.checkpoint.postgres import PostgresSaver # Production checkpointer checkpointer = PostgresSaver.from_conn_string( "postgresql://user:pass@localhost/langgraph" ) agent = create_react_agent(llm, tools, checkpointer=checkpointer) ``` ### Vector Store Memory for Long-Term Context ```python from langchain_community.vectorstores import Chroma from langchain_voyageai import VoyageAIEmbeddings embeddings = VoyageAIEmbeddings(model="voyage-3-large") memory_store = Chroma( collection_name="conversation_memory", embedding_function=embeddings, persist_directory="./memory_db" ) async def retrieve_relevant_memory(query: str, k: int = 5) -> list: """Retrieve relevant past conversations.""" docs = await memory_store.asimilarity_search(query, k=k) return [doc.page_content for doc in docs] async def store_memory(content: str, metadata: dict = {}): """Store conversation in long-term memory.""" await memory_store.aadd_texts([content], metadatas=[metadata]) ``` ## Callback System & LangSmith ### LangSmith Tracing ```python import os from langchain_anthropic import ChatAnthropic # Enable LangSmith tracing os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-api-key" os.environ["LANGCHAIN_PROJECT"] = "my-project" # All LangChain/LangGraph operations are automatically traced llm = ChatAnthropic(model="claude-sonnet-4-6") ``` ### Custom Callback Handler ```python from langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List class CustomCallbackHandler(BaseCallbackHandler): def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs ) -> None: print(f"LLM started with {len(prompts)} prompts") def on_llm_end(self, response, **kwargs) -> None: print(f"LLM completed: {len(response.generations)} generations") def on_llm_error(self, error: Exception, **kwargs) -> None: print(f"LLM error: {error}") def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs ) -> None: print(f"Tool started: {serialized.get('name')}") def on_tool_end(self, output: str, **kwargs) -> None: print(f"Tool completed: {output[:100]}...") # Use callbacks result = await agent.ainvoke( {"messages": [("user", "query")]}, config={"callbacks": [CustomCallbackHandler()]} ) ``` ## Streaming Responses ```python from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-sonnet-4-6", streaming=True) # Stream tokens async for chunk in llm.astream("Tell me a story"): print(chunk.content, end="", flush=True) # Stream agent events async for event in agent.astream_events( {"messages": [("user", "Search and summarize")]}, version="v2" ): if event["event"] == "on_chat_model_stream": print(event["data"]["chunk"].content, end="") elif event["event"] == "on_tool_start": print(f"\n[Using tool: {event['name']}]") ``` ## Testing Strategies ```python import pytest from unittest.mock import AsyncMock, patch @pytest.mark.asyncio async def test_agent_tool_selection(): """Test agent selects correct tool.""" with patch.object(llm, 'ainvoke') as mock_llm: mock_llm.return_value = AsyncMock(content="Using search_database") result = await agent.ainvoke({ "messages": [("user", "search for documents")] }) # Verify tool was called assert "search_database" in str(result) @pytest.mark.asyncio async def test_memory_persistence(): """Test memory persists across invocations.""" config = {"configurable": {"thread_id": "test-thread"}} # First message await agent.ainvoke( {"messages": [("user", "Remember: the code is 12345")]}, config ) # Second message should remember result = await agent.ainvoke( {"messages": [("user", "What was the code?")]}, config ) assert "12345" in result["messages"][-1].content ``` ## Performance Optimization ### 1. Caching with Redis ```python from langchain_community.cache import RedisCache from langchain_core.globals import set_llm_cache import redis redis_client = redis.Redis.from_url("redis://localhost:6379") set_llm_cache(RedisCache(redis_client)) ``` ### 2. Async Batch Processing ```python import asyncio from langchain_core.documents import Document async def process_documents(documents: list[Document]) -> list: """Process documents in parallel.""" tasks = [process_single(doc) for doc in documents] return await asyncio.gather(*tasks) async def process_single(doc: Document) -> dict: """Process a single document.""" chunks = text_splitter.split_documents([doc]) embeddings = await embeddings_model.aembed_documents( [c.page_content for c in chunks] ) return {"doc_id": doc.metadata.get("id"), "embeddings": embeddings} ``` ### 3. Connection Pooling ```python from langchain_pinecone import PineconeVectorStore from pinecone import Pinecone # Reuse Pinecone client pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("my-index") # Create vector store with existing index vectorstore = PineconeVectorStore(index=index, embedding=embeddings) ``` ## Resources - [LangChain Documentation](https://python.langchain.com/docs/) - [LangGraph Documentation](https://langchain-ai.github.io/langgraph/) - [LangSmith Platform](https://smith.langchain.com/) - [LangChain GitHub](https://github.com/langchain-ai/langchain) - [LangGraph GitHub](https://github.com/langchain-ai/langgraph) ## Common Pitfalls 1. **Using Deprecated APIs**: Use LangGraph for agents, not `initialize_agent` 2. **Memory Overflow**: Use checkpointers with TTL for long-running agents 3. **Poor Tool Descriptions**: Clear descriptions help LLM select correct tools 4. **Context Window Exceeded**: Use summarization or sliding window memory 5. **No Error Handling**: Wrap too
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

similarity-search-patterns

Implement efficient similarity search with vector databases. Use

coding
โญ1
# Similarity Search Patterns Patterns for implementing efficient similarity search in production systems. ## When to Use This Skill - Building semantic search systems - Implementing RAG retrieval - Creating recommendation engines - Optimizing search latency - Scaling to millions of vectors - Combining semantic and keyword search ## Core Concepts ### 1. Distance Metrics | Metric | Formula | Best For | | ------------------ | ------------------ | --------------------- | --- | -------------- | | **Cosine** | 1 - (AยทB)/(โ€–Aโ€–โ€–Bโ€–) | Normalized embeddings | | **Euclidean (L2)** | โˆšฮฃ(a-b)ยฒ | Raw embeddings | | **Dot Product** | AยทB | Magnitude matters | | **Manhattan (L1)** | ฮฃ | a-b | | Sparse vectors | ### 2. Index Types ``` โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ” โ”‚ Index Types โ”‚ โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ฌโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”‚ Flat โ”‚ HNSW โ”‚ IVF+PQ โ”‚ โ”‚ (Exact) โ”‚ (Graph-based) โ”‚ (Quantized) โ”‚ โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ผโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ผโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค โ”‚ O(n) search โ”‚ O(log n) โ”‚ O(โˆšn) โ”‚ โ”‚ 100% recall โ”‚ ~95-99% โ”‚ ~90-95% โ”‚ โ”‚ Small data โ”‚ Medium-Large โ”‚ Very Large โ”‚ โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ดโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜ ``` ## Templates ### Template 1: Pinecone Implementation ```python from pinecone import Pinecone, ServerlessSpec from typing import List, Dict, Optional import hashlib class PineconeVectorStore: def __init__( self, api_key: str, index_name: str, dimension: int = 1536, metric: str = "cosine" ): self.pc = Pinecone(api_key=api_key) # Create index if not exists if index_name not in self.pc.list_indexes().names(): self.pc.create_index( name=index_name, dimension=dimension, metric=metric, spec=ServerlessSpec(cloud="aws", region="us-east-1") ) self.index = self.pc.Index(index_name) def upsert( self, vectors: List[Dict], namespace: str = "" ) -> int: """ Upsert vectors. vectors: [{"id": str, "values": List[float], "metadata": dict}] """ # Batch upsert batch_size = 100 total = 0 for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] self.index.upsert(vectors=batch, namespace=namespace) total += len(batch) return total def search( self, query_vector: List[float], top_k: int = 10, namespace: str = "", filter: Optional[Dict] = None, include_metadata: bool = True ) -> List[Dict]: """Search for similar vectors.""" results = self.index.query( vector=query_vector, top_k=top_k, namespace=namespace, filter=filter, include_metadata=include_metadata ) return [ { "id": match.id, "score": match.score, "metadata": match.metadata } for match in results.matches ] def search_with_rerank( self, query: str, query_vector: List[float], top_k: int = 10, rerank_top_n: int = 50, namespace: str = "" ) -> List[Dict]: """Search and rerank results.""" # Over-fetch for reranking initial_results = self.search( query_vector, top_k=rerank_top_n, namespace=namespace ) # Rerank with cross-encoder or LLM reranked = self._rerank(query, initial_results) return reranked[:top_k] def _rerank(self, query: str, results: List[Dict]) -> List[Dict]: """Rerank results using cross-encoder.""" from sentence_transformers import CrossEncoder model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') pairs = [(query, r["metadata"]["text"]) for r in results] scores = model.predict(pairs) for result, score in zip(results, scores): result["rerank_score"] = float(score) return sorted(results, key=lambda x: x["rerank_score"], reverse=True) def delete(self, ids: List[str], namespace: str = ""): """Delete vectors by ID.""" self.index.delete(ids=ids, namespace=namespace) def delete_by_filter(self, filter: Dict, namespace: str = ""): """Delete vectors matching filter.""" self.index.delete(filter=filter, namespace=namespace) ``` ### Template 2: Qdrant Implementation ```python from qdrant_client import QdrantClient from qdrant_client.http import models from typing import List, Dict, Optional class QdrantVectorStore: def __init__( self, url: str = "localhost", port: int = 6333, collection_name: str = "documents", vector_size: int = 1536 ): self.client = QdrantClient(url=url, port=port) self.collection_name = collection_name # Create collection if not exists collections = self.client.get_collections().collections if collection_name not in [c.name for c in collections]: self.client.create_collection( collection_name=collection_name, vectors_config=models.VectorParams( size=vector_size, distance=models.Distance.COSINE ), # Optional: enable quantization for memory efficiency quantization_config=models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=True ) ) ) def upsert(self, points: List[Dict]) -> int: """ Upsert points. points: [{"id": str/int, "vector": List[float], "payload": dict}] """ qdrant_points = [ models.PointStruct( id=p["id"], vector=p["vector"], payload=p.get("payload", {}) ) for p in points ] self.client.upsert( collection_name=self.collection_name, points=qdrant_points ) return len(points) def search( self, query_vector: List[float], limit: int = 10, filter: Optional[models.Filter] = None, score_threshold: Optional[float] = None ) -> List[Dict]: """Search for similar vectors.""" results = self.client.search( collection_name=self.collection_name, query_vector=query_vector, limit=limit, query_filter=filter, score_threshold=score_threshold ) return [ { "id": r.id, "score": r.score, "payload": r.payload } for r in results ] def search_with_filter( self, query_vector: List[float], must_conditions: List[Dict] = None, should_conditions: List[Dict] = None, must_not_conditions: List[Dict] = None, limit: int = 10 ) -> List[Dict]: """Search with complex filters.""" conditions = [] if must_conditions: conditions.extend([ models.FieldCondition( key=c["key"], match=models.MatchValue(value=c["value"]) ) for c in must_conditions ]) filter = models.Filter(must=conditions) if conditions else None return self.search(query_vector, limit=limit, filter=filter) def search_with_sparse( self, dense_vector: List[float], sparse_vector: Dict[int, float], limit: int = 10, dense_weight: float = 0.7 ) -> List[Dict]: """Hybrid search with dense and sparse vectors.""" # Requires collection with named vectors results = self.client.search( collection_name=self.collection_name, query_vector=models.NamedVector( name="dense", vector=dense_vector ), limit=limit ) return [{"id": r.id, "score": r.score, "payload": r.payload} for r in results] ``` ### Template 3: pgvector with PostgreSQL ```python import asyncpg from typing import List, Dict, Optional import numpy as np class PgVectorStore: def __init__(self, connection_string: str): self.connection_string = connection_string async def init(self): """Initialize connection pool and extension.""" self.pool = await asyncpg.create_pool(self.connection_string) async with self.pool.acquire() as conn: # Enable extension await conn.execute("CREATE EXTENSION IF NOT EXISTS vector") # Create table await conn.execute(""" CREATE TABLE IF NOT EXISTS documents ( id TEXT PRIMARY KEY, content TEXT, metadata JSONB, embedding vector(1536) ) """) # Create index (HNSW for better performance) await conn.execute(""" CREATE INDEX IF NOT EXISTS documents_embedding_idx ON documents USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64) """) async def upsert(self, documents: List[Dict]): """Upsert documents with embeddings.""" async with self.pool.acquire() as conn: await conn.executemany( """ INSERT INTO documents (id, content, metadata, embedding) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO UPDATE SET content = EXCLUDED.content, metadata = EXCLUDED.metadata, embedding = EXCLUDED.embedding """, [ ( doc["id"], doc["content"], doc.get("metadata", {}), np.array(doc["embedding"]).tolist() ) for doc in documents ] ) async def search( self, query_embedding: List[float], limit: int = 10, filter_metadata: Optional[Dict] = None ) -> List[Dict]: """Search for similar documents.""" query = """ SELECT id, content, metadata, 1 - (embedding <=> $1::vector) as similarity FROM documents """ params = [query_embedding] if filter_metadata: conditions = [] for key, value in filter_metadata.items(): params.append(value) conditions.append(f"metadata->>'{key}' = ${len(params)}") query += " WHERE " + " AND ".join(conditions) query += f" ORDER BY embedding <=> $1::vector LIMIT ${len(params) + 1}" params.append(limit) async with self.pool.acquire() as conn: rows = await conn.fetch(query, *params) return [ { "id": row["id"], "content": row["content"], "metadata": row["metadata"], "score": row["similarity"] } for row in rows ] async def hybrid_search( self, query_embedding: List[float], query_text: str, limit: int = 10, vector_weight: float = 0.5 ) -> List[Dict]: """Hybrid search combining vector and full-text.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ WITH vector_results AS ( SELECT id, content, metadata, 1 - (embedding <=> $1::vector) as vector_score FROM documents ORDER BY embedding <=> $1::vector LIMIT $3 * 2 ), text_results AS ( SELECT id, content, metadata, ts_rank(to_tsvector('english', content), plainto_tsquery('english', $2)) as text_score FROM documents WHERE to_tsvector('english', content) @@ plainto_tsquery('english', $2) LIMIT $3 * 2 ) SELECT COALESCE(v.id, t.id) as id, COALESCE(v.content, t.content) as content, COALESCE(v.metadata, t.metadata) as metadata, COALESCE(v.vector_score, 0) * $4 + COALESCE(t.text_score, 0) * (1 - $4) as combined_score FROM vector_results v FULL OUTER JOIN text_results t ON v.id = t.id ORDER BY combined_score DESC LIMIT $3 """, query_embedding, query_text, limit, vector_weight ) return [dict(row) for row in rows] ``` ### Template 4: Weaviate Implementation ```python import weaviate from weaviate.util import generate_uuid5 from typing import List, Dict, Optional class WeaviateVectorStore: def __init__( self, url: str = "http://localhost:8080", class_name: str = "Document" ): self.client = weaviate.Client(url=url) self.class_name = class_name self._ensure_schema() def _ensure_schema(self): """Create schema if not exists.""" schema = { "class": self.class_name, "vectorizer": "none", # We provide vectors "properties": [ {"name": "content", "dataType": ["text"]}, {"name": "source", "dataType": ["string"]}, {"name": "chunk_id", "dataType": ["int"]} ] } if not self.client.schema.exists(self.class_name): self.client.schema.create_class(schema) def upsert(self, documents: List[Dict]): """Batch upsert documents.""" with self.client.batch as batch: batch.batch_size = 100 for doc in documents: batch.add_data_object( data_object={ "content": doc["content"], "source": doc.get("source", ""), "chunk_id": doc.get("chunk_id", 0) }, class_name=self.class_name, uuid=generate_uuid5(doc["id"]), vector=doc["embedding"] ) def search( self, query_vector: List[float], limit: int = 10, where_filter: Optional[Dict] = None ) -> List[Dict]: """Vector search.""" query = ( self.client.query .get(self.class_name, ["content", "source", "chunk_id"]) .with_near_vector({"vector": query_vector}) .with_limit(limit) .with_additional(["distance", "id"]) ) if where_filter: query = query.with_where(where_filter) results = query.do() return [ { "id": item["_additional"]["id"], "content": item["content"], "source": item["source"], "score": 1 - item["_additional"]["distance"] } for item in results["data"]["Get"][self.class_name] ] def hybrid_search( self, query: str, query_vector: List[float], limit: int = 10, alpha: float = 0.5 # 0 = keyword, 1 = vector ) -> List[Dict]: """Hybrid search combining BM25 and vector.""" results = ( self.client.query .get(self.class_name, ["content", "source"]) .with_hybrid(query=query, vector=query_vector, alpha=alpha) .with_limit(limit) .with_additional(["score"]) .do() ) return [ { "content": item["content"], "source": item["source"], "score": item["_additional"]["score"] } for item in results["data"]["Get"][self.class_name] ] ``` ## Best Practices ### Do's - **Use appropriate index** - HNSW for most cases - **Tune parameters** - ef_search, nprobe for recall/speed - **Implement hybrid search** - Combine with keyword search - **Monitor recall** - Measure search quality - **Pre-filter when possible** - Reduce search space ### Don'ts - **Don't skip evaluation** - Measure before optimizing - **Don't over-index** - Start with flat, scale up - **Don't ignore latency** - P99 matters for UX - **Don't forget costs** - Vector storage adds up ## Resources - [Pinecone Docs](https://docs.pinecone.io/) - [Qdrant Docs](https://qdrant.tech/documentation/) - [pgvector](https://github.com/pgvector/pgvector) - [Weaviate Docs](https://weaviate.io/developers/weaviate)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

vector-index-tuning

Optimize vector index performance for latency, recall, and memory.

coding
โญ1
# Vector Index Tuning Guide to optimizing vector indexes for production performance. ## When to Use This Skill - Tuning HNSW parameters - Implementing quantization - Optimizing memory usage - Reducing search latency - Balancing recall vs speed - Scaling to billions of vectors ## Core Concepts ### 1. Index Type Selection ``` Data Size Recommended Index โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ < 10K vectors โ†’ Flat (exact search) 10K - 1M โ†’ HNSW 1M - 100M โ†’ HNSW + Quantization > 100M โ†’ IVF + PQ or DiskANN ``` ### 2. HNSW Parameters | Parameter | Default | Effect | | ------------------ | ------- | ---------------------------------------------------- | | **M** | 16 | Connections per node, โ†‘ = better recall, more memory | | **efConstruction** | 100 | Build quality, โ†‘ = better index, slower build | | **efSearch** | 50 | Search quality, โ†‘ = better recall, slower search | ### 3. Quantization Types ``` Full Precision (FP32): 4 bytes ร— dimensions Half Precision (FP16): 2 bytes ร— dimensions INT8 Scalar: 1 byte ร— dimensions Product Quantization: ~32-64 bytes total Binary: dimensions/8 bytes ``` ## Templates ### Template 1: HNSW Parameter Tuning ```python import numpy as np from typing import List, Tuple import time def benchmark_hnsw_parameters( vectors: np.ndarray, queries: np.ndarray, ground_truth: np.ndarray, m_values: List[int] = [8, 16, 32, 64], ef_construction_values: List[int] = [64, 128, 256], ef_search_values: List[int] = [32, 64, 128, 256] ) -> List[dict]: """Benchmark different HNSW configurations.""" import hnswlib results = [] dim = vectors.shape[1] n = vectors.shape[0] for m in m_values: for ef_construction in ef_construction_values: # Build index index = hnswlib.Index(space='cosine', dim=dim) index.init_index(max_elements=n, M=m, ef_construction=ef_construction) build_start = time.time() index.add_items(vectors) build_time = time.time() - build_start # Get memory usage memory_bytes = index.element_count * ( dim * 4 + # Vector storage m * 2 * 4 # Graph edges (approximate) ) for ef_search in ef_search_values: index.set_ef(ef_search) # Measure search search_start = time.time() labels, distances = index.knn_query(queries, k=10) search_time = time.time() - search_start # Calculate recall recall = calculate_recall(labels, ground_truth, k=10) results.append({ "M": m, "ef_construction": ef_construction, "ef_search": ef_search, "build_time_s": build_time, "search_time_ms": search_time * 1000 / len(queries), "recall@10": recall, "memory_mb": memory_bytes / 1024 / 1024 }) return results def calculate_recall(predictions: np.ndarray, ground_truth: np.ndarray, k: int) -> float: """Calculate recall@k.""" correct = 0 for pred, truth in zip(predictions, ground_truth): correct += len(set(pred[:k]) & set(truth[:k])) return correct / (len(predictions) * k) def recommend_hnsw_params( num_vectors: int, target_recall: float = 0.95, max_latency_ms: float = 10, available_memory_gb: float = 8 ) -> dict: """Recommend HNSW parameters based on requirements.""" # Base recommendations if num_vectors < 100_000: m = 16 ef_construction = 100 elif num_vectors < 1_000_000: m = 32 ef_construction = 200 else: m = 48 ef_construction = 256 # Adjust ef_search based on recall target if target_recall >= 0.99: ef_search = 256 elif target_recall >= 0.95: ef_search = 128 else: ef_search = 64 return { "M": m, "ef_construction": ef_construction, "ef_search": ef_search, "notes": f"Estimated for {num_vectors:,} vectors, {target_recall:.0%} recall" } ``` ### Template 2: Quantization Strategies ```python import numpy as np from typing import Optional class VectorQuantizer: """Quantization strategies for vector compression.""" @staticmethod def scalar_quantize_int8( vectors: np.ndarray, min_val: Optional[float] = None, max_val: Optional[float] = None ) -> Tuple[np.ndarray, dict]: """Scalar quantization to INT8.""" if min_val is None: min_val = vectors.min() if max_val is None: max_val = vectors.max() # Scale to 0-255 range scale = 255.0 / (max_val - min_val) quantized = np.clip( np.round((vectors - min_val) * scale), 0, 255 ).astype(np.uint8) params = {"min_val": min_val, "max_val": max_val, "scale": scale} return quantized, params @staticmethod def dequantize_int8( quantized: np.ndarray, params: dict ) -> np.ndarray: """Dequantize INT8 vectors.""" return quantized.astype(np.float32) / params["scale"] + params["min_val"] @staticmethod def product_quantize( vectors: np.ndarray, n_subvectors: int = 8, n_centroids: int = 256 ) -> Tuple[np.ndarray, dict]: """Product quantization for aggressive compression.""" from sklearn.cluster import KMeans n, dim = vectors.shape assert dim % n_subvectors == 0 subvector_dim = dim // n_subvectors codebooks = [] codes = np.zeros((n, n_subvectors), dtype=np.uint8) for i in range(n_subvectors): start = i * subvector_dim end = (i + 1) * subvector_dim subvectors = vectors[:, start:end] kmeans = KMeans(n_clusters=n_centroids, random_state=42) codes[:, i] = kmeans.fit_predict(subvectors) codebooks.append(kmeans.cluster_centers_) params = { "codebooks": codebooks, "n_subvectors": n_subvectors, "subvector_dim": subvector_dim } return codes, params @staticmethod def binary_quantize(vectors: np.ndarray) -> np.ndarray: """Binary quantization (sign of each dimension).""" # Convert to binary: positive = 1, negative = 0 binary = (vectors > 0).astype(np.uint8) # Pack bits into bytes n, dim = vectors.shape packed_dim = (dim + 7) // 8 packed = np.zeros((n, packed_dim), dtype=np.uint8) for i in range(dim): byte_idx = i // 8 bit_idx = i % 8 packed[:, byte_idx] |= (binary[:, i] << bit_idx) return packed def estimate_memory_usage( num_vectors: int, dimensions: int, quantization: str = "fp32", index_type: str = "hnsw", hnsw_m: int = 16 ) -> dict: """Estimate memory usage for different configurations.""" # Vector storage bytes_per_dimension = { "fp32": 4, "fp16": 2, "int8": 1, "pq": 0.05, # Approximate "binary": 0.125 } vector_bytes = num_vectors * dimensions * bytes_per_dimension[quantization] # Index overhead if index_type == "hnsw": # Each node has ~M*2 edges, each edge is 4 bytes (int32) index_bytes = num_vectors * hnsw_m * 2 * 4 elif index_type == "ivf": # Inverted lists + centroids index_bytes = num_vectors * 8 + 65536 * dimensions * 4 else: index_bytes = 0 total_bytes = vector_bytes + index_bytes return { "vector_storage_mb": vector_bytes / 1024 / 1024, "index_overhead_mb": index_bytes / 1024 / 1024, "total_mb": total_bytes / 1024 / 1024, "total_gb": total_bytes / 1024 / 1024 / 1024 } ``` ### Template 3: Qdrant Index Configuration ```python from qdrant_client import QdrantClient from qdrant_client.http import models def create_optimized_collection( client: QdrantClient, collection_name: str, vector_size: int, num_vectors: int, optimize_for: str = "balanced" # "recall", "speed", "memory" ) -> None: """Create collection with optimized settings.""" # HNSW configuration based on optimization target hnsw_configs = { "recall": models.HnswConfigDiff(m=32, ef_construct=256), "speed": models.HnswConfigDiff(m=16, ef_construct=64), "balanced": models.HnswConfigDiff(m=16, ef_construct=128), "memory": models.HnswConfigDiff(m=8, ef_construct=64) } # Quantization configuration quantization_configs = { "recall": None, # No quantization for max recall "speed": models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=True ) ), "balanced": models.ScalarQuantization( scalar=models.ScalarQuantizationConfig( type=models.ScalarType.INT8, quantile=0.99, always_ram=False ) ), "memory": models.ProductQuantization( product=models.ProductQuantizationConfig( compression=models.CompressionRatio.X16, always_ram=False ) ) } # Optimizer configuration optimizer_configs = { "recall": models.OptimizersConfigDiff( indexing_threshold=10000, memmap_threshold=50000 ), "speed": models.OptimizersConfigDiff( indexing_threshold=5000, memmap_threshold=20000 ), "balanced": models.OptimizersConfigDiff( indexing_threshold=20000, memmap_threshold=50000 ), "memory": models.OptimizersConfigDiff( indexing_threshold=50000, memmap_threshold=10000 # Use disk sooner ) } client.create_collection( collection_name=collection_name, vectors_config=models.VectorParams( size=vector_size, distance=models.Distance.COSINE ), hnsw_config=hnsw_configs[optimize_for], quantization_config=quantization_configs[optimize_for], optimizers_config=optimizer_configs[optimize_for] ) def tune_search_parameters( client: QdrantClient, collection_name: str, target_recall: float = 0.95 ) -> dict: """Tune search parameters for target recall.""" # Search parameter recommendations if target_recall >= 0.99: search_params = models.SearchParams( hnsw_ef=256, exact=False, quantization=models.QuantizationSearchParams( ignore=True, # Don't use quantization for search rescore=True ) ) elif target_recall >= 0.95: search_params = models.SearchParams( hnsw_ef=128, exact=False, quantization=models.QuantizationSearchParams( ignore=False, rescore=True, oversampling=2.0 ) ) else: search_params = models.SearchParams( hnsw_ef=64, exact=False, quantization=models.QuantizationSearchParams( ignore=False, rescore=False ) ) return search_params ``` ### Template 4: Performance Monitoring ```python import time from dataclasses import dataclass from typing import List import numpy as np @dataclass class SearchMetrics: latency_p50_ms: float latency_p95_ms: float latency_p99_ms: float recall: float qps: float class VectorSearchMonitor: """Monitor vector search performance.""" def __init__(self, ground_truth_fn=None): self.latencies = [] self.recalls = [] self.ground_truth_fn = ground_truth_fn def measure_search( self, search_fn, query_vectors: np.ndarray, k: int = 10, num_iterations: int = 100 ) -> SearchMetrics: """Benchmark search performance.""" latencies = [] for _ in range(num_iterations): for query in query_vectors: start = time.perf_counter() results = search_fn(query, k=k) latency = (time.perf_counter() - start) * 1000 latencies.append(latency) latencies = np.array(latencies) total_queries = num_iterations * len(query_vectors) total_time = sum(latencies) / 1000 # seconds return SearchMetrics( latency_p50_ms=np.percentile(latencies, 50), latency_p95_ms=np.percentile(latencies, 95), latency_p99_ms=np.percentile(latencies, 99), recall=self._calculate_recall(search_fn, query_vectors, k) if self.ground_truth_fn else 0, qps=total_queries / total_time ) def _calculate_recall(self, search_fn, queries: np.ndarray, k: int) -> float: """Calculate recall against ground truth.""" if not self.ground_truth_fn: return 0 correct = 0 total = 0 for query in queries: predicted = set(search_fn(query, k=k)) actual = set(self.ground_truth_fn(query, k=k)) correct += len(predicted & actual) total += k return correct / total def profile_index_build( build_fn, vectors: np.ndarray, batch_sizes: List[int] = [1000, 10000, 50000] ) -> dict: """Profile index build performance.""" results = {} for batch_size in batch_sizes: times = [] for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] start = time.perf_counter() build_fn(batch) times.append(time.perf_counter() - start) results[batch_size] = { "avg_batch_time_s": np.mean(times), "vectors_per_second": batch_size / np.mean(times) } return results ``` ## Best Practices ### Do's - **Benchmark with real queries** - Synthetic may not represent production - **Monitor recall continuously** - Can degrade with data drift - **Start with defaults** - Tune only when needed - **Use quantization** - Significant memory savings - **Consider tiered storage** - Hot/cold data separation ### Don'ts - **Don't over-optimize early** - Profile first - **Don't ignore build time** - Index updates have cost - **Don't forget reindexing** - Plan for maintenance - **Don't skip warming** - Cold indexes are slow ## Resources - [HNSW Paper](https://arxiv.org/abs/1603.09320) - [Faiss Wiki](https://github.com/facebookresearch/faiss/wiki) - [ANN Benchmarks](https://ann-benchmarks.com/)
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

ml-pipeline-workflow

Build end-to-end MLOps pipelines from data preparation through

coding
โญ1
# ML Pipeline Workflow Complete end-to-end MLOps pipeline orchestration from data preparation through model deployment. ## Overview This skill provides comprehensive guidance for building production ML pipelines that handle the full lifecycle: data ingestion โ†’ preparation โ†’ training โ†’ validation โ†’ deployment โ†’ monitoring. ## When to Use This Skill - Building new ML pipelines from scratch - Designing workflow orchestration for ML systems - Implementing data โ†’ model โ†’ deployment automation - Setting up reproducible training workflows - Creating DAG-based ML orchestration - Integrating ML components into production systems ## What This Skill Provides ### Core Capabilities 1. **Pipeline Architecture** - End-to-end workflow design - DAG orchestration patterns (Airflow, Dagster, Kubeflow) - Component dependencies and data flow - Error handling and retry strategies 2. **Data Preparation** - Data validation and quality checks - Feature engineering pipelines - Data versioning and lineage - Train/validation/test splitting strategies 3. **Model Training** - Training job orchestration - Hyperparameter management - Experiment tracking integration - Distributed training patterns 4. **Model Validation** - Validation frameworks and metrics - A/B testing infrastructure - Performance regression detection - Model comparison workflows 5. **Deployment Automation** - Model serving patterns - Canary deployments - Blue-green deployment strategies - Rollback mechanisms ### Reference Documentation See the `references/` directory for detailed guides: - **data-preparation.md** - Data cleaning, validation, and feature engineering - **model-training.md** - Training workflows and best practices - **model-validation.md** - Validation strategies and metrics - **model-deployment.md** - Deployment patterns and serving architectures ### Assets and Templates The `assets/` directory contains: - **pipeline-dag.yaml.template** - DAG template for workflow orchestration - **training-config.yaml** - Training configuration template - **validation-checklist.md** - Pre-deployment validation checklist ## Usage Patterns ### Basic Pipeline Setup ```python # 1. Define pipeline stages stages = [ "data_ingestion", "data_validation", "feature_engineering", "model_training", "model_validation", "model_deployment" ] # 2. Configure dependencies # See assets/pipeline-dag.yaml.template for full example ``` ### Production Workflow 1. **Data Preparation Phase** - Ingest raw data from sources - Run data quality checks - Apply feature transformations - Version processed datasets 2. **Training Phase** - Load versioned training data - Execute training jobs - Track experiments and metrics - Save trained models 3. **Validation Phase** - Run validation test suite - Compare against baseline - Generate performance reports - Approve for deployment 4. **Deployment Phase** - Package model artifacts - Deploy to serving infrastructure - Configure monitoring - Validate production traffic ## Best Practices ### Pipeline Design - **Modularity**: Each stage should be independently testable - **Idempotency**: Re-running stages should be safe - **Observability**: Log metrics at every stage - **Versioning**: Track data, code, and model versions - **Failure Handling**: Implement retry logic and alerting ### Data Management - Use data validation libraries (Great Expectations, TFX) - Version datasets with DVC or similar tools - Document feature engineering transformations - Maintain data lineage tracking ### Model Operations - Separate training and serving infrastructure - Use model registries (MLflow, Weights & Biases) - Implement gradual rollouts for new models - Monitor model performance drift - Maintain rollback capabilities ### Deployment Strategies - Start with shadow deployments - Use canary releases for validation - Implement A/B testing infrastructure - Set up automated rollback triggers - Monitor latency and throughput ## Integration Points ### Orchestration Tools - **Apache Airflow**: DAG-based workflow orchestration - **Dagster**: Asset-based pipeline orchestration - **Kubeflow Pipelines**: Kubernetes-native ML workflows - **Prefect**: Modern dataflow automation ### Experiment Tracking - MLflow for experiment tracking and model registry - Weights & Biases for visualization and collaboration - TensorBoard for training metrics ### Deployment Platforms - AWS SageMaker for managed ML infrastructure - Google Vertex AI for GCP deployments - Azure ML for Azure cloud - Kubernetes + KServe for cloud-agnostic serving ## Progressive Disclosure Start with the basics and gradually add complexity: 1. **Level 1**: Simple linear pipeline (data โ†’ train โ†’ deploy) 2. **Level 2**: Add validation and monitoring stages 3. **Level 3**: Implement hyperparameter tuning 4. **Level 4**: Add A/B testing and gradual rollouts 5. **Level 5**: Multi-model pipelines with ensemble strategies ## Common Patterns ### Batch Training Pipeline ```yaml # See assets/pipeline-dag.yaml.template stages: - name: data_preparation dependencies: [] - name: model_training dependencies: [data_preparation] - name: model_evaluation dependencies: [model_training] - name: model_deployment dependencies: [model_evaluation] ``` ### Real-time Feature Pipeline ```python # Stream processing for real-time features # Combined with batch training # See references/data-preparation.md ``` ### Continuous Training ```python # Automated retraining on schedule # Triggered by data drift detection # See references/model-training.md ``` ## Troubleshooting ### Common Issues - **Pipeline failures**: Check dependencies and data availability - **Training instability**: Review hyperparameters and data quality - **Deployment issues**: Validate model artifacts and serving config - **Performance degradation**: Monitor data drift and model metrics ### Debugging Steps 1. Check pipeline logs for each stage 2. Validate input/output data at boundaries 3. Test components in isolation 4. Review experiment tracking metrics 5. Inspect model artifacts and metadata ## Next Steps After setting up your pipeline: 1. Explore **hyperparameter-tuning** skill for optimization 2. Learn **experiment-tracking-setup** for MLflow/W&B 3. Review **model-deployment-patterns** for serving strategies 4. Implement monitoring with observability tools ## Related Skills - **experiment-tracking-setup**: MLflow and Weights & Biases integration - **hyperparameter-tuning**: Automated hyperparameter optimization - **model-deployment-patterns**: Advanced deployment strategies
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered
๐Ÿค–system promptโ€ข7 months ago

distributed-tracing

Implement distributed tracing with Jaeger and Tempo to track

coding
โญ1
# Distributed Tracing Implement distributed tracing with Jaeger and Tempo for request flow visibility across microservices. ## Purpose Track requests across distributed systems to understand latency, dependencies, and failure points. ## When to Use - Debug latency issues - Understand service dependencies - Identify bottlenecks - Trace error propagation - Analyze request paths ## Distributed Tracing Concepts ### Trace Structure ``` Trace (Request ID: abc123) โ†“ Span (frontend) [100ms] โ†“ Span (api-gateway) [80ms] โ”œโ†’ Span (auth-service) [10ms] โ””โ†’ Span (user-service) [60ms] โ””โ†’ Span (database) [40ms] ``` ### Key Components - **Trace** - End-to-end request journey - **Span** - Single operation within a trace - **Context** - Metadata propagated between services - **Tags** - Key-value pairs for filtering - **Logs** - Timestamped events within a span ## Jaeger Setup ### Kubernetes Deployment ```bash # Deploy Jaeger Operator kubectl create namespace observability kubectl create -f https://github.com/jaegertracing/jaeger-operator/releases/download/v1.51.0/jaeger-operator.yaml -n observability # Deploy Jaeger instance kubectl apply -f - <<EOF apiVersion: jaegertracing.io/v1 kind: Jaeger metadata: name: jaeger namespace: observability spec: strategy: production storage: type: elasticsearch options: es: server-urls: http://elasticsearch:9200 ingress: enabled: true EOF ``` ### Docker Compose ```yaml version: "3.8" services: jaeger: image: jaegertracing/all-in-one:latest ports: - "5775:5775/udp" - "6831:6831/udp" - "6832:6832/udp" - "5778:5778" - "16686:16686" # UI - "14268:14268" # Collector - "14250:14250" # gRPC - "9411:9411" # Zipkin environment: - COLLECTOR_ZIPKIN_HOST_PORT=:9411 ``` **Reference:** See `references/jaeger-setup.md` ## Application Instrumentation ### OpenTelemetry (Recommended) #### Python (Flask) ```python from opentelemetry import trace from opentelemetry.exporter.jaeger.thrift import JaegerExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.instrumentation.flask import FlaskInstrumentor from flask import Flask # Initialize tracer resource = Resource(attributes={SERVICE_NAME: "my-service"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(JaegerExporter( agent_host_name="jaeger", agent_port=6831, )) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Flask app = Flask(__name__) FlaskInstrumentor().instrument_app(app) @app.route('/api/users') def get_users(): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("get_users") as span: span.set_attribute("user.count", 100) # Business logic users = fetch_users_from_db() return {"users": users} def fetch_users_from_db(): tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("database_query") as span: span.set_attribute("db.system", "postgresql") span.set_attribute("db.statement", "SELECT * FROM users") # Database query return query_database() ``` #### Node.js (Express) ```javascript const { NodeTracerProvider } = require("@opentelemetry/sdk-trace-node"); const { JaegerExporter } = require("@opentelemetry/exporter-jaeger"); const { BatchSpanProcessor } = require("@opentelemetry/sdk-trace-base"); const { registerInstrumentations } = require("@opentelemetry/instrumentation"); const { HttpInstrumentation } = require("@opentelemetry/instrumentation-http"); const { ExpressInstrumentation, } = require("@opentelemetry/instrumentation-express"); // Initialize tracer const provider = new NodeTracerProvider({ resource: { attributes: { "service.name": "my-service" } }, }); const exporter = new JaegerExporter({ endpoint: "http://jaeger:14268/api/traces", }); provider.addSpanProcessor(new BatchSpanProcessor(exporter)); provider.register(); // Instrument libraries registerInstrumentations({ instrumentations: [new HttpInstrumentation(), new ExpressInstrumentation()], }); const express = require("express"); const app = express(); app.get("/api/users", async (req, res) => { const tracer = trace.getTracer("my-service"); const span = tracer.startSpan("get_users"); try { const users = await fetchUsers(); span.setAttributes({ "user.count": users.length }); res.json({ users }); } finally { span.end(); } }); ``` #### Go ```go package main import ( "context" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/exporters/jaeger" "go.opentelemetry.io/otel/sdk/resource" sdktrace "go.opentelemetry.io/otel/sdk/trace" semconv "go.opentelemetry.io/otel/semconv/v1.4.0" ) func initTracer() (*sdktrace.TracerProvider, error) { exporter, err := jaeger.New(jaeger.WithCollectorEndpoint( jaeger.WithEndpoint("http://jaeger:14268/api/traces"), )) if err != nil { return nil, err } tp := sdktrace.NewTracerProvider( sdktrace.WithBatcher(exporter), sdktrace.WithResource(resource.NewWithAttributes( semconv.SchemaURL, semconv.ServiceNameKey.String("my-service"), )), ) otel.SetTracerProvider(tp) return tp, nil } func getUsers(ctx context.Context) ([]User, error) { tracer := otel.Tracer("my-service") ctx, span := tracer.Start(ctx, "get_users") defer span.End() span.SetAttributes(attribute.String("user.filter", "active")) users, err := fetchUsersFromDB(ctx) if err != nil { span.RecordError(err) return nil, err } span.SetAttributes(attribute.Int("user.count", len(users))) return users, nil } ``` **Reference:** See `references/instrumentation.md` ## Context Propagation ### HTTP Headers ``` traceparent: 00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01 tracestate: congo=t61rcWkgMzE ``` ### Propagation in HTTP Requests #### Python ```python from opentelemetry.propagate import inject headers = {} inject(headers) # Injects trace context response = requests.get('http://downstream-service/api', headers=headers) ``` #### Node.js ```javascript const { propagation } = require("@opentelemetry/api"); const headers = {}; propagation.inject(context.active(), headers); axios.get("http://downstream-service/api", { headers }); ``` ## Tempo Setup (Grafana) ### Kubernetes Deployment ```yaml apiVersion: v1 kind: ConfigMap metadata: name: tempo-config data: tempo.yaml: | server: http_listen_port: 3200 distributor: receivers: jaeger: protocols: thrift_http: grpc: otlp: protocols: http: grpc: storage: trace: backend: s3 s3: bucket: tempo-traces endpoint: s3.amazonaws.com querier: frontend_worker: frontend_address: tempo-query-frontend:9095 --- apiVersion: apps/v1 kind: Deployment metadata: name: tempo spec: replicas: 1 template: spec: containers: - name: tempo image: grafana/tempo:latest args: - -config.file=/etc/tempo/tempo.yaml volumeMounts: - name: config mountPath: /etc/tempo volumes: - name: config configMap: name: tempo-config ``` **Reference:** See `assets/jaeger-config.yaml.template` ## Sampling Strategies ### Probabilistic Sampling ```yaml # Sample 1% of traces sampler: type: probabilistic param: 0.01 ``` ### Rate Limiting Sampling ```yaml # Sample max 100 traces per second sampler: type: ratelimiting param: 100 ``` ### Adaptive Sampling ```python from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased # Sample based on trace ID (deterministic) sampler = ParentBased(root=TraceIdRatioBased(0.01)) ``` ## Trace Analysis ### Finding Slow Requests **Jaeger Query:** ``` service=my-service duration > 1s ``` ### Finding Errors **Jaeger Query:** ``` service=my-service error=true tags.http.status_code >= 500 ``` ### Service Dependency Graph Jaeger automatically generates service dependency graphs showing: - Service relationships - Request rates - Error rates - Average latencies ## Best Practices 1. **Sample appropriately** (1-10% in production) 2. **Add meaningful tags** (user_id, request_id) 3. **Propagate context** across all service boundaries 4. **Log exceptions** in spans 5. **Use consistent naming** for operations 6. **Monitor tracing overhead** (<1% CPU impact) 7. **Set up alerts** for trace errors 8. **Implement distributed context** (baggage) 9. **Use span events** for important milestones 10. **Document instrumentation** standards ## Integration with Logging ### Correlated Logs ```python import logging from opentelemetry import trace logger = logging.getLogger(__name__) def process_request(): span = trace.get_current_span() trace_id = span.get_span_context().trace_id logger.info( "Processing request", extra={"trace_id": format(trace_id, '032x')} ) ``` ## Troubleshooting **No traces appearing:** - Check collector endpoint - Verify network connectivity - Check sampling configuration - Review application logs **High latency overhead:** - Reduce sampling rate - Use batch span processor - Check exporter configuration ## Reference Files - `references/jaeger-setup.md` - Jaeger installation - `references/instrumentation.md` - Instrumentation patterns - `assets/jaeger-config.yaml.template` - Jaeger configuration ## Related Skills - `prometheus-configuration` - For metrics - `grafana-dashboards` - For visualization - `slo-implementation` - For latency SLOs
๐Ÿ‘0
๐Ÿ‘๏ธ0
๐Ÿค– Auto-discovered