0
# Watchtower
1

2
A lightweight Python logging handler that integrates the Python logging system with Amazon Web Services CloudWatch Logs. Watchtower provides seamless log streaming to CloudWatch without requiring system-wide log collectors, using the boto3 AWS SDK for efficient log transmission with batching, delivery guarantees, and built-in filtering to prevent infinite logging loops from AWS SDK dependencies.
3

4
## Package Information
5

6
- **Package Name**: watchtower
7
- **Language**: Python
8
- **Installation**: `pip install watchtower`
9
- **Dependencies**: boto3 >= 1.9.253, < 2
10

11
## Core Imports
12

13
```python
14
import watchtower
15
```
16

17
Common pattern for logging setup:
18

19
```python
20
from watchtower import CloudWatchLogHandler, CloudWatchLogFormatter, WatchtowerError, WatchtowerWarning
21
```
22

23
For type annotations:
24

25
```python
26
from typing import Any, Callable, Dict, List, Optional, Tuple
27
import botocore.client
28
import logging
29
```
30

31
## Basic Usage
32

33
```python
34
import watchtower
35
import logging
36

37
# Basic setup
38
logging.basicConfig(level=logging.INFO)
39
logger = logging.getLogger(__name__)
40
logger.addHandler(watchtower.CloudWatchLogHandler())
41

42
# Simple logging
43
logger.info("Hi")
44
logger.error("Something went wrong")
45

46
# Structured logging with JSON
47
logger.info(dict(foo="bar", details={"status": "success", "count": 42}))
48
```
49

50
## Advanced Configuration
51

52
```python
53
import boto3
54
import watchtower
55

56
# Custom configuration
57
handler = watchtower.CloudWatchLogHandler(
58
    log_group_name="my-application",
59
    log_stream_name="{machine_name}/{program_name}/{logger_name}",
60
    use_queues=True,
61
    send_interval=30,
62
    max_batch_size=512 * 1024,
63
    boto3_client=boto3.client("logs", region_name="us-west-2"),
64
    create_log_group=True,
65
    log_group_tags={"Environment": "production", "Service": "web-api"}
66
)
67

68
logger = logging.getLogger("myapp")
69
logger.addHandler(handler)
70
logger.info("Application started")
71
```
72

73
## Architecture
74

75
Watchtower uses a thread-based queuing system for efficient log delivery:
76

77
- **CloudWatchLogHandler**: Main handler that implements Python's logging.Handler interface
78
- **Message Queuing**: Thread-safe queue system that batches messages for efficient delivery
79
- **Batch Processing**: Automatic batching based on size, count, and time intervals
80
- **AWS Integration**: Direct boto3 CloudWatch Logs API integration with retry logic
81
- **Stream Management**: Automatic log group and stream creation with naming templates
82

83
## Capabilities
84

85
### CloudWatch Log Handler
86

87
The main handler class that integrates with Python's logging system to send logs to AWS CloudWatch Logs. Provides comprehensive configuration options for log grouping, streaming, batching, and AWS authentication.
88

89
**Class Constants:**
90
```python { .api }
91
END: int = 1  # Signal for thread termination
92
FLUSH: int = 2  # Signal for immediate flush
93
FLUSH_TIMEOUT: int = 30  # Timeout for flush operations
94
EXTRA_MSG_PAYLOAD_SIZE: int = 26  # Extra metadata size per message
95
```
96

97
```python { .api }
98
class CloudWatchLogHandler(logging.Handler):
99
    def __init__(
100
        self,
101
        log_group_name: str = __name__,
102
        log_stream_name: str = "{machine_name}/{program_name}/{logger_name}/{process_id}",
103
        use_queues: bool = True,
104
        send_interval: int = 60,
105
        max_batch_size: int = 1024 * 1024,
106
        max_batch_count: int = 10000,
107
        boto3_client: Optional[botocore.client.BaseClient] = None,
108
        boto3_profile_name: Optional[str] = None,
109
        create_log_group: bool = True,
110
        log_group_tags: Dict[str, str] = {},
111
        json_serialize_default: Optional[Callable] = None,
112
        log_group_retention_days: Optional[int] = None,
113
        create_log_stream: bool = True,
114
        max_message_size: int = 256 * 1024,
115
        log_group: Optional[str] = None,  # deprecated
116
        stream_name: Optional[str] = None,  # deprecated
117
        *args,
118
        **kwargs
119
    ):
120
        """
121
        Create a CloudWatch log handler.
122

123
        Parameters:
124
        - log_group_name: CloudWatch log group name
125
        - log_stream_name: Log stream name template with format placeholders
126
        - use_queues: Enable message queuing and batching (recommended: True)
127
        - send_interval: Maximum seconds to hold messages before sending
128
        - max_batch_size: Maximum batch size in bytes (AWS limit: 1,048,576)
129
        - max_batch_count: Maximum messages per batch (AWS limit: 10,000)
130
        - boto3_client: Custom boto3 logs client (botocore.client.BaseClient) for authentication/region
131
        - boto3_profile_name: AWS profile name for authentication
132
        - create_log_group: Auto-create log group if it doesn't exist
133
        - log_group_tags: Dictionary of tags to apply to log group
134
        - json_serialize_default: Custom JSON serialization function
135
        - log_group_retention_days: Log retention period in days
136
        - create_log_stream: Auto-create log stream if it doesn't exist
137
        - max_message_size: Maximum message size in bytes (default: 256KB)
138
        - log_group: (deprecated) Use log_group_name instead
139
        - stream_name: (deprecated) Use log_stream_name instead
140
        """
141

142
    def emit(self, record: logging.LogRecord) -> None:
143
        """Send a log record to CloudWatch Logs."""
144

145
    def flush(self) -> None:
146
        """Send any queued messages to CloudWatch immediately."""
147

148
    def close(self) -> None:
149
        """Send queued messages and prevent further processing."""
150

151
    def __repr__(self) -> str:
152
        """Return string representation of the handler."""
153
```
154

155
### CloudWatch Log Formatter
156

157
Specialized formatter that handles JSON serialization for CloudWatch Logs, enabling structured logging with automatic recognition and indexing by CloudWatch.
158

159
```python { .api }
160
class CloudWatchLogFormatter(logging.Formatter):
161
    add_log_record_attrs: Tuple[str, ...] = tuple()
162
    
163
    def __init__(
164
        self,
165
        *args,
166
        json_serialize_default: Optional[Callable] = None,
167
        add_log_record_attrs: Optional[Tuple[str, ...]] = None,
168
        **kwargs
169
    ):
170
        """
171
        Create a CloudWatch log formatter.
172

173
        Parameters:
174
        - json_serialize_default: Custom JSON serialization function for objects
175
        - add_log_record_attrs: Tuple of LogRecord attributes to include in messages
176
        """
177

178
    def format(self, record: logging.LogRecord) -> str:
179
        """
180
        Format log record for CloudWatch, handling JSON serialization.
181

182
        Parameters:
183
        - record: LogRecord instance to format
184

185
        Returns:
186
        str: Formatted log message, JSON string for dict messages
187
        """
188
```
189

190
### Exceptions and Warnings
191

192
Custom exception and warning classes for watchtower-specific error handling.
193

194
```python { .api }
195
class WatchtowerError(Exception):
196
    """Default exception class for watchtower module errors."""
197

198
class WatchtowerWarning(UserWarning):
199
    """Default warning class for watchtower module warnings."""
200
```
201

202
### Constants and Utility Functions
203

204
```python { .api }
205
DEFAULT_LOG_STREAM_NAME: str = "{machine_name}/{program_name}/{logger_name}/{process_id}"
206

207
def _json_serialize_default(o: Any) -> str:
208
    """
209
    Standard JSON serializer function for CloudWatch log messages.
210
    
211
    Serializes datetime objects using .isoformat() method,
212
    and all other objects using repr().
213
    
214
    Parameters:
215
    - o: Object to serialize
216
    
217
    Returns:
218
    str: Serialized representation
219
    """
220
```
221

222
Default log stream name template with format placeholders for:
223
- `{machine_name}`: Platform hostname
224
- `{program_name}`: Program name from sys.argv[0]
225
- `{logger_name}`: Logger name
226
- `{process_id}`: Process ID
227
- `{thread_name}`: Thread name
228
- `{strftime:%format}`: UTC datetime formatting
229

230
## Usage Examples
231

232
### Flask Integration
233

234
```python
235
import watchtower
236
import flask
237
import logging
238

239
logging.basicConfig(level=logging.INFO)
240
app = flask.Flask("myapp")
241
handler = watchtower.CloudWatchLogHandler(log_group_name=app.name)
242
app.logger.addHandler(handler)
243
logging.getLogger("werkzeug").addHandler(handler)
244

245
@app.route('/')
246
def hello():
247
    app.logger.info("Request received", extra={"user_agent": request.headers.get("User-Agent")})
248
    return 'Hello World!'
249
```
250

251
### Django Integration
252

253
```python
254
# In settings.py
255
import boto3
256

257
LOGGING = {
258
    'version': 1,
259
    'disable_existing_loggers': False,
260
    'handlers': {
261
        'watchtower': {
262
            'class': 'watchtower.CloudWatchLogHandler',
263
            'log_group_name': 'django-app',
264
            'boto3_client': boto3.client('logs', region_name='us-east-1'),
265
            'create_log_group': True,
266
        },
267
    },
268
    'loggers': {
269
        'django': {
270
            'handlers': ['watchtower'],
271
            'level': 'INFO',
272
        },
273
    },
274
}
275
```
276

277
### Structured Logging with Metadata
278

279
```python
280
import watchtower
281
import logging
282
from datetime import datetime
283

284
# Configure handler with LogRecord attributes
285
handler = watchtower.CloudWatchLogHandler()
286
handler.formatter.add_log_record_attrs = ["levelname", "filename", "process", "thread"]
287

288
logger = logging.getLogger("myapp")
289
logger.addHandler(handler)
290

291
# Log structured data
292
logger.critical({
293
    "event": "user_login",
294
    "user_id": "12345",
295
    "timestamp": datetime.utcnow(),
296
    "metadata": {
297
        "ip_address": "192.168.1.1",
298
        "user_agent": "Mozilla/5.0...",
299
        "session_id": "abc123"
300
    }
301
})
302
```
303

304
### Custom JSON Serialization
305

306
```python
307
import watchtower
308
import logging
309
from datetime import datetime
310
from decimal import Decimal
311

312
def custom_serializer(obj):
313
    """Custom JSON serializer for special types."""
314
    if isinstance(obj, Decimal):
315
        return float(obj)
316
    elif isinstance(obj, datetime):
317
        return obj.isoformat()
318
    return str(obj)
319

320
handler = watchtower.CloudWatchLogHandler(
321
    json_serialize_default=custom_serializer
322
)
323

324
logger = logging.getLogger("myapp")
325
logger.addHandler(handler)
326

327
# Log with custom types
328
logger.info({
329
    "price": Decimal("19.99"),
330
    "timestamp": datetime.utcnow(),
331
    "status": "success"
332
})
333
```
334

335
### Manual Flush and Close
336

337
```python
338
import watchtower
339
import logging
340

341
handler = watchtower.CloudWatchLogHandler()
342
logger = logging.getLogger("myapp")
343
logger.addHandler(handler)
344

345
try:
346
    logger.info("Starting process")
347
    # ... application logic ...
348
    logger.info("Process completed")
349
finally:
350
    # Ensure all logs are sent before exit
351
    handler.flush()  # Send queued messages
352
    handler.close()  # Send final messages and shutdown
353
```
354

355
## Error Handling
356

357
### Common Warnings
358

359
```python
360
# Empty message warning
361
logger.info("")  # Triggers WatchtowerWarning
362

363
# Message after shutdown warning
364
handler.close()
365
logger.info("This triggers a warning")  # WatchtowerWarning
366

367
# Oversized message warning
368
logger.info("x" * (300 * 1024))  # Message truncated, WatchtowerWarning
369
```
370

371
### AWS Authentication Errors
372

373
```python
374
import watchtower
375
import boto3
376
from botocore.exceptions import NoCredentialsError, ClientError
377

378
try:
379
    handler = watchtower.CloudWatchLogHandler(
380
        boto3_client=boto3.client("logs", region_name="us-west-2")
381
    )
382
except NoCredentialsError:
383
    print("AWS credentials not configured")
384
except ClientError as e:
385
    print(f"AWS API error: {e}")
386
```
387

388
### Configuration Validation
389

390
```python
391
import watchtower
392

393
try:
394
    # This raises WatchtowerError
395
    handler = watchtower.CloudWatchLogHandler(
396
        boto3_client=boto3.client("logs"),
397
        boto3_profile_name="myprofile"  # Can't specify both
398
    )
399
except watchtower.WatchtowerError as e:
400
    print(f"Configuration error: {e}")
401
```