Coding it my way

Block Storage vs Object Storage

2026-04-01T00:00:00+05:30

In this article, we will look at Block Storage and Object Storage and differences between them.

Block Storage

Block storage consists of Block servers whose responsibility is to take a file, split it into blocks, then optionally compress and encrypt it and then finally store it into actual storage.

One of the advantage Block servers provide is Delta sync. When a file is modified, only modified blocks are synced instead of the whole file using a sync algorithm.

Block servers process filed passed from clients by splitting a file into blocks, compressing each block, and encrypting them.

Question-> How do they determine which blocks are modified?

Frequent modification Block storage supports frequent data writes without affecting performance. Instead of rewriting > the entire file, the system identifies the particular block that needs to be amended. Then, it rewrites the selected block with the new data. This makes block storage very efficient for managing large files that require frequent updates.

Example of block storage is Amazon EBS

Block storage gives the flexibility of storing frequent data on SSD while non-frequent data on HDD.

How does it work?

Block servers process files passed from clients by splitting a file into blocks, compress each block and encrypting them. Instead of uploading the whole file to the storage system, only modified blocks are transferred.

The advantage of it is that in version history, you need not store the entire file. Only modified blocks.

Block storage offers high performance, low latency, and quick data transfer rates. As it operates on a block level, you can directly access data and achieve a high I/O performance. You use block storage for applications that need fast access to data you have stored, like a virtual machine or database.

Object Storage

Object storage is different from Block storage

Object storage systems prioritize storage quantity over availability. As highly scalable systems, you can store large volume of unstructured data in an object storage system. However, there’s more latency when you access these files. Object storage also has a lower throughput compared to block storage and cloud storage.

Example of object storage is Amazon s3

Block Storage vs Object Storage

Block storage is mainly useful for structured data with high volumes of read and write loads.

Th key here is that in these files, mostly you don’t need the entire file in its entirety, but only the part of it. For example few rows of database, few files inside the VM mount etc.

Block Storage is like having a filing cabinet with lots of small, numbered drawers. Perfect when you neet to quickly grab a specific drawer.

Object storage is best used for large amouts of unstructured data.

In object storage, you typically need the whole file to operate. for example Images, Videos, PDF, backups etc.

Object Storage is like having a warehouse with labeled boxes. Perfect when you need to retrieve entire boxes (complete files)

References: https://aws.amazon.com/compare/the-difference-between-block-file-object-storage/

Systemd timer running on every boot

2026-03-01T00:00:00+05:30

Disclaimer: This blog article has been generated with the assistance of AI. While the content is AI-generated, the software itself and the ideas behind it are the result of real development work and genuine user needs.

Problem Description

A systemd timer configured with OnCalendar=daily and Persistent=true was executing its associated service on every system boot, in addition to the scheduled execution time. The expected behavior was execution only at the scheduled time, with persistent timers catching up on missed runs only when the system was offline during the scheduled time.

Environment

System: User systemd instance (systemctl --user)
Timer: rednotebook-backup.timer
Service: rednotebook-backup.service
Systemd version: Standard systemd installation

Observed Behavior

Aug 13 20:59:24 - Service executed after boot
-- Boot f6bac27bcb034b92807c437ec2834916 --
Aug 13 21:10:50 - Service executed after boot
-- Boot 18a831a9d19b415da72eb836e78ad866 --
Aug 13 21:11:20 - Service executed after boot

The service executed immediately after each boot, regardless of the last execution time or scheduled interval.

Configuration Analysis

Original Timer Configuration

[Unit]
Description=Run RedNotebook backup daily
Requires=rednotebook-backup.service

[Timer]
OnCalendar=daily
Persistent=true
RandomizedDelaySec=300

[Install]
WantedBy=timers.target

Service Configuration

[Unit]
Description=RedNotebook GitHub Backup Service
After=network-online.target 
Wants=network-online.target

[Service]
Type=oneshot
ExecStart=%h/bin/rednotebook-backup.sh
ExecStartPre=/bin/sleep 30
Environment=HOME=%h
StandardOutput=journal
StandardError=journal
Environment=SSH_AUTH_SOCK=%t/ssh-agent.socket

Root Cause Analysis

The issue stems from the Requires=rednotebook-backup.service directive in the timer unit’s [Unit] section.

Systemd Dependency Resolution

Timer activation: The timer unit is started by timers.target during boot
Dependency resolution: Due to Requires=rednotebook-backup.service, systemd immediately activates the service unit when the timer starts
Unintended execution: The service executes on boot, independent of the timer schedule

Systemd Timer Mechanics

Systemd timers automatically establish relationships with their target services through:

Implicit naming convention: foo.timer automatically targets foo.service
Automatic dependencies: Timer units gain an implicit Before= dependency on their target service
No explicit linking required: The Unit= directive in [Timer] section is only needed when targeting a service with a different name

From systemd.timer(5):

Timer units automatically gain a Before= dependency on the service they are supposed to activate.

Solution

Remove the Requires= directive from the timer unit:

[Unit]
Description=Run RedNotebook backup daily
# Removed: Requires=rednotebook-backup.service

[Timer]
OnCalendar=daily
Persistent=true
RandomizedDelaySec=300

[Install]
WantedBy=timers.target

Implementation

systemctl --user stop rednotebook-backup.timer
systemctl --user edit rednotebook-backup.timer
# Add override: [Unit]\nRequires=
systemctl --user daemon-reload
systemctl --user start rednotebook-backup.timer

Verification

Post-fix timer state:

NEXT                            LEFT     LAST PASSED UNIT                     ACTIVATES                 
Thu 2025-08-14 00:02:17 IST    2h 54min  -    -      rednotebook-backup.timer rednotebook-backup.service

Multiple reboot test results:

Service does not execute on boot
Timer maintains correct scheduling
Service executes only at scheduled intervals

Common Misconceptions

Explicit linking requirement: Many assume timers require explicit service dependencies
Cargo-cult configuration: Copy-pasting examples with unnecessary Requires= directives
Documentation gaps: Some tutorials incorrectly show Requires= as standard practice

Alternative Diagnostic Approaches

If removing Requires= doesn’t resolve the issue:

Check service enablement:
```
systemctl --user is-enabled rednotebook-backup.service
```
Services triggered by timers should show static or disabled

Verify service dependencies:

systemctl --user list-dependencies rednotebook-backup.service

Check for WantedBy directives: Remove WantedBy= from service [Install] section

Timer State Management

Systemd maintains timer state in:

System timers: /var/lib/systemd/timers/
User timers: ~/.local/share/systemd/

State corruption can cause similar symptoms but is less common than configuration issues.

Technical References

systemd.timer(5) - Timer unit configuration
systemd.unit(5) - Unit configuration directives
systemd.special(7) - Special systemd units
systemd GitHub Issue #14642 - Timer runs on every boot

Conclusion

The Requires= directive in systemd timer units creates an unnecessary dependency that causes service execution on timer activation (boot). Systemd’s implicit timer-to-service binding mechanism eliminates the need for explicit dependencies in standard configurations. This issue affects both user and system timer units and represents a common configuration anti-pattern in systemd timer implementations.

Displaying Google Calendar Events in Conky

2026-02-01T00:00:00+05:30

Disclaimer: This blog article has been generated with the assistance of AI. While the content is AI-generated, the software itself and the ideas behind it are the result of real development work and genuine user needs.

Disclaimer: This article was generated with assistance from AI and reflects a technical implementation discussion.

Conky is a popular system monitor for Linux desktops, but its capabilities extend beyond displaying CPU usage and memory stats. This guide shows how to integrate Google Calendar events into your Conky display using Python and Google’s Calendar API, with proper support for recurring events.

Why Use the API Instead of iCal?

While Google Calendar provides iCal feeds, they have significant limitations for recurring events:

iCal feeds only include expanded instances for a limited time window
Future recurring events beyond that window aren’t included
No control over the expansion period

The Google Calendar API with singleEvents=true parameter properly expands recurring events for any specified time range.

Prerequisites

Python 3.8+
pipenv (for dependency management)
A Google Cloud Console account
Conky installed on your Linux system

Setup Process

1. Create Service Account

Navigate to Google Cloud Console:

Create or select a project
Enable the Google Calendar API (APIs & Services → Library)
Create a Service Account (IAM & Admin → Service Accounts)
Generate a JSON key file for the service account
Download and save as google_calendar_key.json

2. Grant Calendar Access

The service account needs permission to read your calendar:

Open the JSON file and copy the client_email value
In Google Calendar, go to Settings → Your Calendar → Share with specific people
Add the service account email with “See all event details” permission

3. Project Structure

mkdir -p ~/.conky/calendar-project
cd ~/.conky/calendar-project

Create Pipfile:

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
python-dotenv = "*"
google-auth = "*"
google-auth-oauthlib = "*"
google-auth-httplib2 = "*"
google-api-python-client = "*"

[dev-packages]

[requires]
python_version = "3.8"

Install dependencies:

pipenv install

4. Configuration

Create .env file:

GOOGLE_SERVICE_ACCOUNT_FILE=google_calendar_key.json
GOOGLE_CALENDAR_ID=primary
MAX_EVENTS=10
CACHE_DAYS=7

Set secure permissions:

chmod 600 .env google_calendar_key.json

5. Python Script

Create get_calendar.py:

#!/usr/bin/env python3
import os
import socket
from datetime import datetime, timedelta
from pathlib import Path

try:
    from dotenv import load_dotenv
    from google.oauth2 import service_account
    from googleapiclient.discovery import build
    DEPENDENCIES_AVAILABLE = True
except ImportError as e:
    DEPENDENCIES_AVAILABLE = False
    print(f"Missing dependencies: {e}")

# Load configuration
script_dir = Path(__file__).parent
env_file = script_dir / '.env'
if env_file.exists():
    load_dotenv(env_file)

SERVICE_ACCOUNT_FILE = os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE', 'google_calendar_key.json')
CALENDAR_ID = os.getenv('GOOGLE_CALENDAR_ID', 'primary')
MAX_EVENTS = int(os.getenv('MAX_EVENTS', '10'))
CACHE_DAYS = int(os.getenv('CACHE_DAYS', '7'))
CACHE_FILE = os.path.expanduser("~/.conky/calendar_cache.txt")
SCOPES = ['https://www.googleapis.com/auth/calendar.readonly']

def check_internet():
    try:
        socket.create_connection(("8.8.8.8", 53), timeout=3)
        return True
    except OSError:
        return False

def get_calendar_service():
    try:
        script_dir = Path(__file__).parent
        service_account_path = script_dir / SERVICE_ACCOUNT_FILE
        
        if not service_account_path.exists():
            raise FileNotFoundError(f"Service account file not found: {service_account_path}")
        
        credentials = service_account.Credentials.from_service_account_file(
            str(service_account_path), scopes=SCOPES
        )
        return build('calendar', 'v3', credentials=credentials)
    except Exception as e:
        print(f"Error creating calendar service: {e}")
        return None

def get_events():
    try:
        service = get_calendar_service()
        if not service:
            return None
        
        now = datetime.now()
        start_time = now.replace(hour=0, minute=0, second=0, microsecond=0)
        end_time = start_time + timedelta(days=CACHE_DAYS)
        
        time_min = start_time.isoformat() + 'Z'
        time_max = end_time.isoformat() + 'Z'
        
        # Key parameter: singleEvents=True expands recurring events
        events_result = service.events().list(
            calendarId=CALENDAR_ID,
            timeMin=time_min,
            timeMax=time_max,
            maxResults=MAX_EVENTS * 2,
            singleEvents=True,  # Expands recurring events
            orderBy='startTime'
        ).execute()
        
        events = events_result.get('items', [])
        return events[:MAX_EVENTS]
    except Exception as e:
        print(f"Error fetching events: {e}")
        return None

def format_event(event):
    try:
        summary = event.get('summary', 'No Title')
        start = event.get('start', {})
        
        if 'dateTime' in start:
            dt_str = start['dateTime']
            dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00')).astimezone()
            time_str = dt.strftime('%m/%d %H:%M' if CACHE_DAYS > 1 else '%H:%M')
        elif 'date' in start:
            date_obj = datetime.fromisoformat(start['date'])
            time_str = date_obj.strftime('%m/%d All Day' if CACHE_DAYS > 1 else 'All Day')
        else:
            time_str = '??:??'
        
        # Truncate long summaries
        max_length = 35 if CACHE_DAYS > 1 else 40
        if len(summary) > max_length:
            summary = summary[:max_length-3] + '...'
        
        # Add recurring indicator
        if event.get('recurringEventId'):
            summary += " ↻"
        
        return f"{time_str} - {summary}"
    except Exception as e:
        return f"Error formatting event: {e}"

def load_cache():
    try:
        if os.path.exists(CACHE_FILE):
            with open(CACHE_FILE, 'r') as f:
                return f.read().strip()
    except Exception:
        pass
    return None

def save_cache(content):
    try:
        cache_dir = os.path.dirname(CACHE_FILE)
        os.makedirs(cache_dir, exist_ok=True)
        with open(CACHE_FILE, 'w') as f:
            f.write(content)
    except Exception as e:
        print(f"Error saving cache: {e}")

def main():
    if not DEPENDENCIES_AVAILABLE:
        print("Required dependencies not installed")
        return
    
    script_dir = Path(__file__).parent
    if not (script_dir / SERVICE_ACCOUNT_FILE).exists():
        print(f"Service account file not found: {SERVICE_ACCOUNT_FILE}")
        return
    
    if not check_internet():
        cached_content = load_cache()
        if cached_content:
            print(cached_content)
        else:
            print("No internet connection and no cached data")
        return
    
    events = get_events()
    
    if events is None:
        cached_content = load_cache()
        if cached_content:
            print("Using cached data:")
            print(cached_content)
        else:
            print("Unable to fetch calendar events")
        return
    
    if not events:
        days_text = f"next {CACHE_DAYS} day{'s' if CACHE_DAYS > 1 else ''}" if CACHE_DAYS > 1 else "today"
        output = f"No events {days_text}"
    else:
        formatted_events = [format_event(event) for event in events]
        output = '\n'.join(formatted_events)
    
    save_cache(output)
    print(output)

if __name__ == '__main__':
    main()

Make executable:

chmod +x get_calendar.py

6. Conky Integration

Add to your .conkyrc:

conky.text = [[
${color orange}${font DejaVu Sans Mono:bold:size=12}UPCOMING EVENTS${font}${color}
${hr 2}
${color lightblue}${exec cd ~/.conky/calendar-project && pipenv run python get_calendar.py}${color}
${hr 1}
]]

Testing

cd ~/.conky/calendar-project
pipenv run python get_calendar.py

Expected output:

08/12 22:00 - Zendesk Redis Withdraw
08/16 08:00 - Weekly Reflection... ↻
08/19 All Day - Credit card bill ↻

Key Features

Recurring Event Support: The ↻ symbol indicates recurring events
Offline Caching: Falls back to cached data when internet is unavailable
Configurable Time Range: Adjust CACHE_DAYS for different periods
Timezone Awareness: Properly handles timezone conversions
Error Handling: Graceful degradation on API failures

Configuration Options

Variable	Description	Example
`CACHE_DAYS`	Days to look ahead	`7` (one week)
`MAX_EVENTS`	Maximum events to display	`10`
`CALENDAR_ID`	Which calendar to use	`primary` or email

Troubleshooting

“Service account file not found”

Verify the JSON file is in the project directory
Check the filename matches your .env configuration

“Access denied”

Ensure the service account email has calendar permissions
Wait a few minutes after sharing for propagation

“No events found”

Verify CALENDAR_ID is correct
Check if events exist in the specified time range
Try increasing CACHE_DAYS

Security Considerations

Service account files contain sensitive credentials
Use chmod 600 on credential files
Consider restricting API key to specific APIs in Google Cloud Console
The service account only needs read access to calendars

Limitations

Requires internet connectivity for fresh data (cached data available offline)
Service account must be shared with each calendar individually
Google Calendar API has usage quotas (generous for personal use)

This implementation provides a robust solution for displaying calendar events in Conky with proper support for recurring events and offline functionality.

Image location detection using Bedrock and Claude Model

2026-01-01T00:00:00+05:30

Disclaimer: This blog article has been generated with the assistance of AI. While the content is AI-generated, the software itself and the ideas behind it are the result of real development work and genuine user needs.

I needed a way to automatically identify locations in a collection of images, so I built a Python script that uses AWS Bedrock’s Claude API for image analysis. The script processes images in batches, optimizes them for API efficiency, and saves location descriptions as text files.

Overview

The script handles:

Recursive image processing from folders
Image optimization to reduce API costs
AWS Bedrock integration with Claude models
Automatic text file generation for results
Skip logic for already-processed images

Setup

Install dependencies:

pip install boto3 Pillow

Configure AWS credentials via CLI, environment variables, or IAM roles. Ensure you have access to Bedrock and Claude models in your region.

Image Processing

The first challenge is managing image sizes. Large images are expensive to process and may exceed API limits. The solution processes images entirely in memory:

import os
import base64
from PIL import Image
import io

def resize_image_in_memory(image_path, max_size=(1024, 1024), quality=85):
    """Resize image in memory without saving to disk"""
    try:
        with Image.open(image_path) as img:
            # Convert to RGB if necessary (handles RGBA, P mode images)
            if img.mode in ('RGBA', 'P'):
                img = img.convert('RGB')
            
            # Calculate new size maintaining aspect ratio
            img.thumbnail(max_size, Image.Resampling.LANCZOS)
            
            # Save to bytes buffer
            buffer = io.BytesIO()
            
            # Determine format based on original image
            original_format = img.format if img.format else 'JPEG'
            if original_format.upper() in ['JPEG', 'JPG']:
                img.save(buffer, format='JPEG', quality=quality, optimize=True)
            elif original_format.upper() == 'PNG':
                img.save(buffer, format='PNG', optimize=True)
            else:
                # Default to JPEG for other formats
                img.save(buffer, format='JPEG', quality=quality, optimize=True)
            
            buffer.seek(0)
            return buffer.getvalue(), original_format
            
    except Exception as e:
        print(f"Error resizing image {image_path}: {e}")
        return None, None

This maintains aspect ratios, handles different formats, and provides significant compression. For a typical 4MB photo, this reduces size to ~300KB (92% reduction).

Base64 Encoding

AWS Bedrock requires base64-encoded images. The encoding function includes optional optimization:

def encode_image_to_base64(image_path, resize=True, max_size=(1024, 1024), quality=85):
    """Encode image file to base64 string with optional resizing"""
    try:
        if resize:
            image_bytes, original_format = resize_image_in_memory(image_path, max_size, quality)
            if image_bytes is None:
                return None
            return base64.b64encode(image_bytes).decode('utf-8')
        else:
            with open(image_path, "rb") as image_file:
                return base64.b64encode(image_file.read()).decode('utf-8')
    except Exception as e:
        print(f"Error encoding image {image_path}: {e}")
        return None

Bedrock Integration

The core function handles the API communication:

import boto3
import mimetypes
from botocore.exceptions import ClientError

def get_image_format(image_path):
    """Get image format for Bedrock API (lowercase format names)"""
    mime_type, _ = mimetypes.guess_type(image_path)
    if mime_type and mime_type.startswith('image/'):
        format_map = {
            'image/jpeg': 'jpeg',
            'image/jpg': 'jpeg', 
            'image/png': 'png',
            'image/gif': 'gif',
            'image/webp': 'webp'
        }
        return format_map.get(mime_type, 'jpeg')
    return 'jpeg'

def analyze_image_location(bedrock_client, image_path, model_id="anthropic.claude-3-5-sonnet-20241022-v2:0", 
                         resize_images=True, max_size=(1024, 1024), quality=85):
    """Send image to AWS Bedrock and get location analysis"""
    try:
        base64_image = encode_image_to_base64(image_path, resize=resize_images, 
                                            max_size=max_size, quality=quality)
        if not base64_image:
            return None
        
        image_format = get_image_format(image_path)
        
        message = {
            "role": "user",
            "content": [
                {
                    "text": "Analyze this image and identify the location or place shown. Provide a concise one-line description of the location, including the most specific place name possible (city, landmark, or geographical feature). Keep the response to a single sentence - no more than one line. Do not start with phrases like 'This is', 'The image shows', or 'This appears to be'"
                },
                {
                    "image": {
                        "format": image_format,
                        "source": {
                            "bytes": base64.b64decode(base64_image)
                        }
                    }
                }
            ]
        }
        
        response = bedrock_client.converse(
            modelId=model_id,
            messages=[message],
            inferenceConfig={
                "maxTokens": 1000,
                "temperature": 0.1
            }
        )
        
        location_analysis = response['output']['message']['content'][0]['text']
        return location_analysis
        
    except ClientError as e:
        print(f"AWS Bedrock error for {image_path}: {e}")
        return None
    except Exception as e:
        print(f"Error analyzing image {image_path}: {e}")
        return None

Key technical details:

Uses the converse API instead of the older invoke method
Low temperature (0.1) for consistent responses
Specific prompt engineering to avoid verbose prefixes
Handles various image formats through MIME type detection

File Management

The script organizes results by saving text files alongside images:

from pathlib import Path

def save_location_to_file(image_path, location_text):
    """Save location analysis to text file with same name as image"""
    try:
        image_name = Path(image_path).stem
        image_dir = Path(image_path).parent
        text_file_path = image_dir / f"{image_name}.txt"
        
        with open(text_file_path, 'w', encoding='utf-8') as f:
            f.write(location_text)
        
        print(f"  Saved to: {text_file_path.name}")
        return True
        
    except Exception as e:
        print(f"Error saving text file for {image_path}: {e}")
        return False

def get_image_files(folder_path):
    """Recursively get all image files from folder and subfolders"""
    image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'}
    image_files = []
    
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            file_path = Path(root) / file
            if file_path.suffix.lower() in image_extensions:
                image_files.append(file_path)
    
    return image_files

Main Processing Loop

The main function coordinates everything:

def process_images(folder_path, aws_region='us-east-1', model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
                  resize_images=True, max_size=(1024, 1024), quality=85):
    """Main function to process all images in folder"""
    
    try:
        bedrock_client = boto3.client(
            service_name='bedrock-runtime',
            region_name=aws_region
        )
        print(f"Connected to AWS Bedrock in region: {aws_region}")
    except Exception as e:
        print(f"Error connecting to AWS Bedrock: {e}")
        return
    
    image_files = get_image_files(folder_path)
    
    if not image_files:
        print(f"No image files found in {folder_path}")
        return
    
    print(f"Found {len(image_files)} image files to process")
    
    processed = 0
    failed = 0
    
    for i, image_path in enumerate(image_files, 1):
        print(f"\n[{i}/{len(image_files)}] Processing: {image_path.name}")
        
        # Skip if already processed
        text_file_path = image_path.parent / f"{image_path.stem}.txt"
        if text_file_path.exists():
            print(f"  Text file already exists, skipping...")
            continue
        
        location_analysis = analyze_image_location(
            bedrock_client, image_path, model_id,
            resize_images=resize_images, max_size=max_size, quality=quality
        )
        
        if location_analysis:
            if save_location_to_file(image_path, location_analysis):
                processed += 1
                print(f"  ✓ Success")
            else:
                failed += 1
                print(f"  ✗ Failed to save")
        else:
            failed += 1
            print(f"  ✗ Failed to analyze")
    
    print(f"\nProcessing Complete: {processed} successful, {failed} failed")

Usage

Configure and run:

if __name__ == "__main__":
    RESIZE_IMAGES = True
    MAX_SIZE = (1024, 1024)
    QUALITY = 85
    
    process_images(
        folder_path="/home/user/Pictures/wallpapers/night",
        aws_region="ap-south-1",
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        resize_images=RESIZE_IMAGES,
        max_size=MAX_SIZE,
        quality=QUALITY
    )

Output Example

Connected to AWS Bedrock in region: ap-south-1
Found 15 image files to process

[1/15] Processing: sunset_beach.jpg
  Size: 4.2MB → 0.3MB (92.8% reduction)
  Saved to: sunset_beach.txt
  ✓ Success

[2/15] Processing: mountain_view.png
  Text file already exists, skipping...

Technical Notes

Model Options:

claude-3-5-sonnet: Good balance of speed and accuracy
claude-3-haiku: Faster and cheaper
claude-3-opus: Most capable but expensive

Error Handling: The script handles network issues, API limits, invalid formats, and file permissions gracefully.

Cost Considerations: Image resizing typically reduces API costs by 80-95%. The skip logic prevents reprocessing already-analyzed images.

Automating Gmail PDF Extraction and Paperless NGX Integration

2025-12-29T00:00:00+05:30

Managing financial documents like bank statements, credit card bills, and tax documents can be tedious. I’ve automated this process with a Python script that:

Scans Gmail for emails with PDF attachments
Extracts and decrypts password-protected PDFs
Organizes files by date and subject
Uploads documents to Paperless NGX for long-term storage and OCR
Sends notifications via ntfy

This script runs daily via a systemd timer, ensuring I never miss important documents. All processed documents are accessible through the Paperless NGX web interface and mobile app, making it easy to access important documents on the go.

Features

Automatic PDF Extraction: Scans Gmail for emails matching configured subject patterns
Password Removal: Automatically decrypts password-protected PDFs using configurable patterns
Email-to-PDF Conversion: Converts entire emails (not just attachments) to PDF for newsletters and important emails
Smart Organization: Organizes files by date (Month-Year) and email subject
Paperless NGX Integration: Automatically uploads processed documents with metadata
Google Drive Backup: Optional Google Drive upload for additional backup
Notifications: Sends notifications via ntfy about processing status
Duplicate Detection: Skips files that already exist

Architecture

The script consists of several modules:

gmail_reader.py: Main script that orchestrates the entire process
pdf_processor.py: Handles PDF decryption and validation
paperless_uploader.py: Manages Paperless NGX API integration
google_drive_uploader.py: Handles Google Drive uploads (optional)
ntfy_notifier.py: Sends push notifications

Installation

Prerequisites

pip install PyPDF2 weasyprint html2text pyyaml requests google-auth google-auth-oauthlib google-auth-httplib2 google-api-python-client

Configuration

Create a config.yaml file:

documents:
  Bank Statement:
    subject_pattern: "Your Bank Statement"
    password_pattern: "your_password_here"
    
  Credit Card:
    subject_pattern: "Your Credit Card Statement"
    password_pattern: "card_number@{ddmm}"  # Date-based password
    
  Tax Document:
    subject_pattern: "Your ITR"
    password_pattern: "pan_number@dob"

# Email-to-PDF conversion for newsletters
email_to_pdf:
  newsletters:
    enabled: true
    subject_regex: "Monthly Newsletter"
    allow_duplicates: false

# Email filters
email_filters:
  max_results: 50
  days_to_search: 30

# PDF settings
pdf_settings:
  output_directory: "decrypted_statements"
  open_after_decrypt: false

# Gmail settings
email_settings:
  email: "your_email@gmail.com"
  app_password: "your_app_password"  # Generate from Google Account settings

# Notifications (optional)
ntfy_settings:
  enabled: true
  server_url: "https://ntfy.sh"  # Or your own ntfy server
  topic: "your_topic_name"
  default_priority: "default"

# Google Drive (optional)
google_drive_settings:
  enabled: false
  service_account_file: "oauth_credentials.json"
  upload_after_decrypt: true
  keep_local_copy: true

# Paperless NGX settings
paperless_settings:
  enabled: true
  base_url: "https://paperless.example.com"
  api_token: "your_api_token"  # Get from Paperless admin panel
  timeout: 30
  default_tags:
    - "Email Import"
    - "Auto Processed"
  default_correspondent: "Email System"
  default_document_type: "Statement"

Gmail App Password

To use Gmail IMAP, you need to generate an app password:

Go to Google Account Settings
Navigate to Security → 2-Step Verification
Scroll down to “App passwords”
Generate a new app password for “Mail”
Use this password in your config.yaml

Core Script

Here’s the main gmail_reader.py script:

import os
import re
import yaml
import email
import imaplib
import argparse
from datetime import datetime, timedelta
from base64 import b64decode
from google_drive_uploader import upload_to_drive
from paperless_uploader import upload_to_paperless
import logging
from typing import List, Dict, Any, Tuple, Optional

from pdf_processor import PDFProcessor
from ntfy_notifier import NtfyNotifier

# For email-to-PDF conversion
import weasyprint
from email.message import EmailMessage
import html2text
import html
from email.utils import parsedate_to_datetime


class ImprovedGmailPDFProcessor:
    def __init__(self, config_path='config.yaml'):
        """Initialize with configuration file."""
        self.setup_logging()
        self.load_config(config_path)
        self.setup_output_directory()
        self.pdf_processor = PDFProcessor()
        self.notifier = NtfyNotifier(self.config.get('ntfy_settings', {}))
        
        # Track processing statistics
        self.stats = {
            'total_emails_found': 0,
            'total_pdfs_found': 0,
            'new_pdfs_processed': 0,
            'skipped_existing': 0,
            'errors': 0,
            'paperless_uploads': 0,
            'paperless_failures': 0,
            'email_pdfs_created': 0,
            'email_pdf_failures': 0,
            'new_files': []
        }

    def setup_logging(self):
        """Setup logging configuration."""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('gmail_pdf_processor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)

    def load_config(self, config_path):
        """Load configuration from YAML file."""
        try:
            with open(config_path, 'r') as config_file:
                self.config = yaml.safe_load(config_file)
            self.logger.info("Configuration loaded successfully")
        except Exception as e:
            error_msg = f"Error loading config file: {e}"
            self.logger.error(error_msg)
            raise Exception(error_msg)

    def connect_to_gmail(self) -> imaplib.IMAP4_SSL:
        """Connect to Gmail using IMAP."""
        try:
            mail = imaplib.IMAP4_SSL("imap.gmail.com")
            mail.login(self.config['email_settings']['email'], 
                      self.config['email_settings']['app_password'])
            self.logger.info("Successfully connected to Gmail")
            return mail
        except Exception as e:
            error_msg = f"Failed to connect to Gmail: {e}"
            self.logger.error(error_msg)
            raise Exception(error_msg)

    def get_recent_emails_for_pattern(self, mail: imaplib.IMAP4_SSL, subject_pattern: str) -> List[bytes]:
        """Fetch recent emails matching a specific pattern."""
        mail.select('INBOX')
        
        days = self.config['email_filters']['days_to_search']
        date_since = (datetime.now() - timedelta(days=days)).strftime("%d-%b-%Y")
        search_criteria = f'(SINCE "{date_since}" SUBJECT "{subject_pattern}")'
        
        self.logger.info(f"Searching for pattern '{subject_pattern}' since {date_since}")
        
        result, messages = mail.search(None, search_criteria)
        messages = messages[0].split() if messages[0] else []
        
        max_results = self.config['email_filters']['max_results']
        return messages[-max_results:] if messages else []

    def is_pdf_attachment(self, part) -> bool:
        """Check if email part is a PDF attachment."""
        content_type = part.get_content_type()
        filename = part.get_filename()
        
        if content_type == 'application/pdf':
            return True
        
        if content_type == 'application/octet-stream' and filename:
            return filename.lower().endswith('.pdf')
        
        content_disposition = part.get('Content-Disposition', '')
        if 'attachment' in content_disposition and filename:
            return filename.lower().endswith('.pdf')
        
        return False

    def get_email_date(self, email_message) -> str:
        """Extract email date from message."""
        try:
            date_str = email_message['Date']
            if not date_str:
                self.logger.warning("Email has no Date header, using current date")
                return datetime.now().strftime("%Y/%m/%d")
            
            email_date = parsedate_to_datetime(date_str)
            return email_date.strftime("%Y/%m/%d")
            
        except Exception as e:
            self.logger.error(f"Error parsing email date: {e}")
            return datetime.now().strftime("%Y/%m/%d")

    def create_date_subfolder(self, email_date_str: str, base_output_dir: str) -> str:
        """Create date-based subfolder (Month-YYYY format)."""
        try:
            email_date = datetime.strptime(email_date_str, "%Y/%m/%d")
            folder_name = email_date.strftime("%B-%Y")
            subfolder_path = os.path.join(base_output_dir, folder_name)
            
            if not os.path.exists(subfolder_path):
                os.makedirs(subfolder_path)
                self.logger.info(f"Created subfolder: {subfolder_path}")
            
            return subfolder_path
            
        except Exception as e:
            self.logger.error(f"Error creating date subfolder: {e}")
            return base_output_dir

    def sanitize_folder_name(self, folder_name: str) -> str:
        """Sanitize folder name by removing invalid characters."""
        invalid_chars = '<>:"/\\|?*'
        for char in invalid_chars:
            folder_name = folder_name.replace(char, '_')
        
        folder_name = folder_name.strip(' .')
        
        if len(folder_name) > 100:
            folder_name = folder_name[:100]
        
        if not folder_name:
            folder_name = "Unknown_Subject"
            
        return folder_name

    def get_password_for_statement(self, doc_config: Dict[str, Any], email_date: str) -> str:
        """Generate password based on pattern and date."""
        password_pattern = doc_config['password_pattern']
        date = datetime.strptime(email_date, "%Y/%m/%d")
        
        replacements = {
            '{yyyy}': date.strftime('%Y'),
            '{yy}': date.strftime('%y'),
            '{mm}': date.strftime('%m'),
            '{dd}': date.strftime('%d'),
            '{ddmm}': date.strftime('%d%m'),
            '{mmyy}': date.strftime('%m%y'),
        }
        
        for placeholder, value in replacements.items():
            password_pattern = password_pattern.replace(placeholder, value)
        
        return password_pattern

    def process_email(self, mail: imaplib.IMAP4_SSL, message_id: bytes, 
                     doc_type: str, doc_config: Dict[str, Any]) -> None:
        """Process a single email message."""
        try:
            _, msg_data = mail.fetch(message_id, '(RFC822)')
            email_body = msg_data[0][1]
            email_message = email.message_from_bytes(email_body)
            
            subject = self.get_email_subject(email_message)
            if not subject:
                return

            subject_normalized = ' '.join(subject.split())
            if re.search(doc_config['subject_pattern'], subject_normalized, re.IGNORECASE):
                self.logger.info(f"Found matching email for {doc_type}: {subject}")
                self.stats['total_emails_found'] += 1
                
                email_date = self.get_email_date(email_message)
                base_output_dir = self.config['pdf_settings']['output_directory']
                date_subfolder = self.create_date_subfolder(email_date, base_output_dir)
                subject_folder = self.create_subject_folder(date_subfolder, subject)
                
                # Process attachments
                for part in email_message.walk():
                    if self.is_pdf_attachment(part):
                        filename = part.get_filename()
                        if filename:
                            self.stats['total_pdfs_found'] += 1
                            
                            output_filename = os.path.join(subject_folder, filename)
                            
                            if self.pdf_processor.check_existing_pdf(output_filename):
                                self.logger.info(f"PDF already exists, skipping: {output_filename}")
                                self.stats['skipped_existing'] += 1
                                continue
                            
                            pdf_data = part.get_payload(decode=True)
                            
                            if not self.pdf_processor.is_pdf_data_valid(pdf_data):
                                self.logger.warning(f"Invalid PDF data for {filename}")
                                self.stats['errors'] += 1
                                continue
                            
                            password = self.get_password_for_statement(doc_config, email_date)
                            
                            success, error_msg = self.pdf_processor.remove_password_protection(
                                pdf_data, password, output_filename
                            )
                            
                            if success:
                                self.logger.info(f"Successfully processed new PDF: {output_filename}")
                                self.stats['new_pdfs_processed'] += 1
                                self.stats['new_files'].append({
                                    'filename': filename,
                                    'path': output_filename,
                                    'doc_type': doc_type,
                                    'subject': subject
                                })
                                
                                # Upload to Paperless NGX if configured
                                if self.config.get('paperless_settings', {}).get('enabled', False):
                                    try:
                                        paperless_success = upload_to_paperless(output_filename, self.config, doc_type)
                                        if paperless_success:
                                            self.stats['paperless_uploads'] += 1
                                        else:
                                            self.stats['paperless_failures'] += 1
                                    except Exception as e:
                                        self.stats['paperless_failures'] += 1
                                        self.logger.error(f"Error uploading to Paperless NGX: {e}")
                                    
        except Exception as e:
            self.logger.error(f"Error processing email {message_id}: {e}")
            self.stats['errors'] += 1

    def run(self) -> None:
        """Main execution function."""
        try:
            self.logger.info("Starting the improved PDF processor...")
            
            mail = self.connect_to_gmail()
            
            # Process each document type
            for doc_type, doc_config in self.config.get('documents', {}).items():
                self.logger.info(f"Processing {doc_type} documents...")
                messages = self.get_recent_emails_for_pattern(mail, doc_config['subject_pattern'])
                self.logger.info(f"Found {len(messages)} matching emails for {doc_type}")
                
                for i, message_id in enumerate(messages, 1):
                    self.logger.info(f"Processing {doc_type} email {i} of {len(messages)}")
                    self.process_email(mail, message_id, doc_type, doc_config)
            
            mail.logout()
            
            self.print_processing_summary()
            self.send_completion_notification()
            
            self.logger.info("Processing completed successfully!")
                
        except Exception as e:
            error_msg = f"An error occurred: {e}"
            self.logger.error(error_msg)
            self.notifier.send_error_notification(
                message=f"PDF processor failed: {str(e)[:200]}...",
                title="PDF Processor Error"
            )
            raise


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='Gmail PDF Processor')
    parser.add_argument('-c', '--config', type=str, default='config.yaml',
                       help='Path to the configuration YAML file')
    args = parser.parse_args()
    
    if not os.path.exists(args.config):
        print(f"Error: Configuration file '{args.config}' not found!")
        exit(1)
    
    processor = ImprovedGmailPDFProcessor(config_path=args.config)
    processor.run()

PDF Processor Module

The pdf_processor.py handles PDF decryption:

import os
import tempfile
import PyPDF2
import logging
from typing import Tuple, Optional


class PDFProcessor:
    """Handles PDF processing operations like password removal."""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def is_pdf_data_valid(self, pdf_data: bytes) -> bool:
        """Check if the provided data is valid PDF content."""
        if not pdf_data:
            return False
        return pdf_data.startswith(b'%PDF')
    
    def check_existing_pdf(self, file_path: str) -> bool:
        """Check if a PDF file already exists at the given path."""
        if not os.path.exists(file_path):
            return False
        
        try:
            with open(file_path, 'rb') as f:
                header = f.read(8)
                if header.startswith(b'%PDF'):
                    return True
                return False
        except Exception as e:
            self.logger.warning(f"Error checking existing PDF {file_path}: {e}")
            return False
    
    def remove_password_protection(self, 
                                  pdf_data: bytes, 
                                  password: str, 
                                  output_path: str) -> Tuple[bool, Optional[str]]:
        """Remove password protection from PDF and save it."""
        temp_path = None
        
        try:
            if not self.is_pdf_data_valid(pdf_data):
                return False, "Invalid PDF data provided"
            
            with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_file:
                temp_file.write(pdf_data)
                temp_path = temp_file.name
            
            pdf_reader = PyPDF2.PdfReader(temp_path)
            
            if pdf_reader.is_encrypted:
                if not pdf_reader.decrypt(password):
                    return False, f"Failed to decrypt PDF with provided password"
                self.logger.info("Successfully decrypted password-protected PDF")
            else:
                self.logger.info("PDF was not password-protected")
            
            pdf_writer = PyPDF2.PdfWriter()
            
            for page in pdf_reader.pages:
                pdf_writer.add_page(page)
            
            os.makedirs(os.path.dirname(output_path), exist_ok=True)
            
            with open(output_path, 'wb') as output_file:
                pdf_writer.write(output_file)
            
            self.logger.info(f"Successfully processed PDF: {output_path}")
            return True, None
            
        except Exception as e:
            error_msg = f"Error processing PDF: {e}"
            self.logger.error(error_msg)
            return False, error_msg
            
        finally:
            if temp_path and os.path.exists(temp_path):
                try:
                    os.unlink(temp_path)
                except Exception as e:
                    self.logger.warning(f"Failed to delete temporary file {temp_path}: {e}")

Paperless NGX Uploader

The paperless_uploader.py handles integration with Paperless NGX:

import os
import requests
import logging
from datetime import datetime
from typing import Optional, Dict, Any, List
import mimetypes


class PaperlessNGXUploader:
    """A generic uploader for Paperless NGX instances."""
    
    def __init__(self, config: Dict[str, Any]):
        self.base_url = config['base_url'].rstrip('/')
        self.api_token = config['api_token']
        self.timeout = config.get('timeout', 30)
        
        self.headers = {
            'Authorization': f'Token {self.api_token}',
            'Accept': 'application/json'
        }
        
        self.default_tags = config.get('default_tags', [])
        self.default_correspondent = config.get('default_correspondent')
        self.default_document_type = config.get('default_document_type')
        
        self.logger = logging.getLogger(__name__)
        
        self._tag_cache = {}
        self._correspondent_cache = {}
        self._document_type_cache = {}

    def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        """Make an API request to Paperless NGX."""
        url = f"{self.base_url}/api{endpoint}"
        
        try:
            response = requests.request(
                method=method,
                url=url,
                headers=self.headers,
                timeout=self.timeout,
                **kwargs
            )
            response.raise_for_status()
            return response
        except requests.exceptions.RequestException as e:
            self.logger.error(f"API request failed: {e}")
            raise

    def _get_or_create_tag(self, tag_name: str) -> int:
        """Get tag ID by name, create if doesn't exist."""
        if tag_name in self._tag_cache:
            return self._tag_cache[tag_name]
        
        response = self._make_request('GET', '/tags/', params={'name': tag_name})
        results = response.json()['results']
        
        if results:
            tag_id = results[0]['id']
        else:
            data = {'name': tag_name}
            response = self._make_request('POST', '/tags/', json=data)
            tag_id = response.json()['id']
            self.logger.info(f"Created new tag: {tag_name} (ID: {tag_id})")
        
        self._tag_cache[tag_name] = tag_id
        return tag_id

    def upload_file(self, 
                   file_path: str,
                   title: Optional[str] = None,
                   tags: Optional[List[str]] = None,
                   correspondent: Optional[str] = None,
                   document_type: Optional[str] = None,
                   created_date: Optional[datetime] = None) -> Dict[str, Any]:
        """Upload a file to Paperless NGX."""
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"File not found: {file_path}")
        
        filename = os.path.basename(file_path)
        self.logger.info(f"Uploading file to Paperless NGX: {filename}")
        
        mime_type, _ = mimetypes.guess_type(file_path)
        if mime_type is None:
            mime_type = 'application/octet-stream'
        
        form_data = {}
        
        if title:
            form_data['title'] = title
        else:
            form_data['title'] = os.path.splitext(filename)[0]
        
        tag_ids = []
        all_tags = (tags or []) + self.default_tags
        for tag_name in all_tags:
            if tag_name:
                tag_id = self._get_or_create_tag(tag_name)
                tag_ids.append(tag_id)
        
        if tag_ids:
            form_data['tags'] = tag_ids
        
        correspondent_name = correspondent or self.default_correspondent
        if correspondent_name:
            correspondent_id = self._get_or_create_correspondent(correspondent_name)
            form_data['correspondent'] = correspondent_id
        
        doc_type_name = document_type or self.default_document_type
        if doc_type_name:
            doc_type_id = self._get_or_create_document_type(doc_type_name)
            form_data['document_type'] = doc_type_id
        
        if created_date:
            form_data['created'] = created_date.strftime('%Y-%m-%d')
        
        with open(file_path, 'rb') as file:
            files = {
                'document': (filename, file, mime_type)
            }
            
            response = self._make_request(
                'POST',
                '/documents/post_document/',
                data=form_data,
                files=files
            )
        
        result = response.json()
        self.logger.info(f"Successfully uploaded {filename} to Paperless NGX")
        return result

    def test_connection(self) -> bool:
        """Test connection to Paperless NGX instance."""
        try:
            response = self._make_request('GET', '/documents/', params={'page_size': 1})
            self.logger.info("Successfully connected to Paperless NGX")
            return True
        except Exception as e:
            self.logger.error(f"Failed to connect to Paperless NGX: {e}")
            return False

Running Paperless NGX in Docker

Paperless NGX is a document management system that automatically OCRs and indexes your documents. Here’s how to run it using Docker Compose:

Docker Compose Configuration

Create a docker-compose.yml file:

services:
  db:
    image: docker.io/library/postgres:17
    restart: unless-stopped
    volumes:
      - pgdata:/var/lib/postgresql/data
    environment:
      POSTGRES_DB: paperless
      POSTGRES_USER: paperless
      POSTGRES_PASSWORD: paperless
    networks:
      - paperless-network

  webserver:
    image: ghcr.io/paperless-ngx/paperless-ngx:latest
    restart: unless-stopped
    depends_on:
      - db
      - gotenberg
      - tika
    volumes:
      - data:/usr/src/paperless/data
      - media:/usr/src/paperless/media
      - ./export:/usr/src/paperless/export
      - ./consume:/usr/src/paperless/consume
    env_file: docker-compose.env
    networks:
      - paperless-network
    environment:
      PAPERLESS_DBHOST: db
      PAPERLESS_TIKA_ENABLED: 1
      PAPERLESS_TIKA_GOTENBERG_ENDPOINT: http://gotenberg:3000
      PAPERLESS_TIKA_ENDPOINT: http://tika:9998

  gotenberg:
    image: docker.io/gotenberg/gotenberg:8.22
    restart: unless-stopped
    networks:
      - paperless-network
    command:
      - "gotenberg"
      - "--chromium-disable-javascript=true"
      - "--chromium-allow-list=file:///tmp/.*"

  tika:
    image: docker.io/apache/tika:latest
    networks:
      - paperless-network
    restart: unless-stopped

volumes:
  data:
  media:
  pgdata:

networks:
  paperless-network:
    driver: bridge

Environment Configuration

Create a docker-compose.env file:

# Paperless URL (required for public access)
PAPERLESS_URL=https://paperless.example.com

# Secret key (generate a long random string)
PAPERLESS_SECRET_KEY=your_secret_key_here

# Timezone
PAPERLESS_TIME_ZONE=America/New_York

# OCR Language
PAPERLESS_OCR_LANGUAGE=eng

Installation Steps

Create directories:

mkdir -p paperless/{export,consume}
cd paperless

Create the docker-compose.yml and docker-compose.env files (as shown above)
Pull images:
```
docker compose pull
```
Start Paperless:
```
docker compose up -d
```

Create admin user:

docker compose exec webserver createsuperuser

Get API token:
- Log in to Paperless web interface
- Go to Settings → API Tokens
- Create a new token
- Use this token in your config.yaml

Accessing Paperless

Web interface: http://localhost:8000 (or your configured URL)
API documentation: http://localhost:8000/api/docs/

Mobile Access with Paperless NGX Android App

One of the best features of this setup is the ability to access all your documents on the go using the Paperless Mobile app by Anton Stubenbord. This is a native Android app (not just a webview) that provides an excellent mobile experience for managing your Paperless NGX documents.

Features of the mobile app:

Full document access: View and search all your processed documents, including bank statements, credit card bills, and tax documents
Document management: Add, delete, or edit documents directly from your phone
Share and download: Share, download, print and preview documents
Scan and upload: Scan documents with your phone camera and upload them with preset correspondent, document type, tags and creation date
Inbox management: Review and quickly process newly added documents in the inbox
Security: Secure your data with biometric authentication
Multiple instances: Seamlessly switch between different accounts and Paperless instances
Modern UI: Built according to Material Design 3 specification with light and dark themes
Multi-language support: Available in English, German, French, Spanish, Catalan, Polish, Czech, Russian and Turkish

Setting up the mobile app:

Install the Paperless Mobile app from Google Play Store
Open the app and add your Paperless NGX server URL
Log in with your credentials (or use API token authentication)
Start browsing your documents!

The app has a 4.6-star rating with over 50K+ downloads and provides a native mobile experience that’s much better than using the web interface on mobile.

Here’s a screenshot of the app showing my processed documents:

The mobile app makes it incredibly convenient to access important documents like bank statements or tax documents when you’re away from your computer. Since all documents are automatically OCR’d by Paperless NGX, you can search for specific text within documents directly from your phone.

Automation with Systemd

To run the script daily, create a systemd timer:

Service File (`~/.config/systemd/user/gmail-pdf-processor.service`)

[Unit]
Description=Gmail PDF Processor
After=network.target

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /path/to/gmail_reader.py -c /path/to/config.yaml
WorkingDirectory=/path/to/email_reader
StandardOutput=journal
StandardError=journal

Timer File (`~/.config/systemd/user/gmail-pdf-processor.timer`)

[Unit]
Description=Run Gmail PDF Processor daily
Requires=gmail-pdf-processor.service

[Timer]
OnCalendar=daily
OnCalendar=*-*-* 09:00:00
Persistent=true

[Install]
WantedBy=timers.target

Enable and Start

systemctl --user daemon-reload
systemctl --user enable gmail-pdf-processor.timer
systemctl --user start gmail-pdf-processor.timer
systemctl --user status gmail-pdf-processor.timer

Benefits

Zero Manual Work: Documents are automatically extracted, decrypted, and organized
Searchable Archive: Paperless NGX OCRs all documents, making them fully searchable
Mobile Access: Access all your documents on the go with the Paperless NGX Android app
Backup: Documents are stored locally and optionally backed up to Google Drive
Notifications: Get notified about new documents via push notifications
Organization: Files are automatically organized by date and subject

Security Considerations

Store your config.yaml with proper permissions (chmod 600)
Use app passwords instead of your main Gmail password
Keep your Paperless NGX instance behind a VPN or reverse proxy with authentication
Regularly update dependencies for security patches

Conclusion

This automation has saved me countless hours of manual document management. The combination of Gmail scanning, PDF decryption, and Paperless NGX creates a seamless document management pipeline that runs completely unattended.

The code is modular and extensible, making it easy to add new document types or integrate with other services. Feel free to adapt it to your needs!

Building a RAG System for Your Jekyll Blog with AWS Bedrock Claude

2025-12-01T00:00:00+05:30

Disclaimer: This blog article has been generated with the assistance of AI. While the content is AI-generated, the software itself and the ideas behind it are the result of real development work and genuine user needs.

Head over to the Complete source code on GitHub

Ever wanted to create a smart assistant that can answer questions specifically about your blog content? In this post, I’ll walk you through building a Retrieval Augmented Generation (RAG) system that indexes all your Jekyll blog posts in a vector database and uses AWS Bedrock Claude to provide intelligent answers based solely on your content.

What We’re Building

Our RAG system will:

✅ Automatically discover and parse Jekyll blog posts
✅ Create semantic embeddings of blog content
✅ Store embeddings in a FAISS vector database
✅ Use AWS Bedrock Claude to answer questions
✅ Ensure responses are based only on blog content
✅ Provide source citations for transparency

Architecture Overview

Jekyll Blog Posts → Content Processing → Embeddings → Vector DB (FAISS)
                                                           ↓
User Question → Query Embeddings → Similarity Search → Context Retrieval
                                                           ↓
Context + Question → AWS Bedrock Claude → Contextualized Answer

Why This Approach Works

Traditional Search Limitations: Standard blog search relies on keyword matching, missing semantic relationships and context.

RAG Advantages:

Semantic Understanding: Vector embeddings capture meaning, not just keywords
Context-Aware Responses: Claude understands nuanced technical concepts
Source Attribution: Always cites which blog posts informed the answer
Content Accuracy: Responses are grounded in your actual writing

Prerequisites

Before we dive in, ensure you have:

A Jekyll blog with markdown posts
Python 3.8+ environment
AWS account with Bedrock access
Basic familiarity with vector databases

Core Components Explained

1. Content Discovery and Parsing

The system automatically finds Jekyll posts in common locations:

post_patterns = [
    self.blog_path / "_posts" / "*.md",
    self.blog_path / "_posts" / "*.markdown",
    self.blog_path / "blog" / "*.md",
    self.blog_path / "posts" / "*.md",
]

It then parses the frontmatter and content, extracting metadata like titles, dates, tags, and categories while cleaning the markdown for better indexing.

2. Content Chunking Strategy

Long blog posts are split into overlapping chunks to improve retrieval accuracy:

def chunk_content(self, content: str, max_tokens: int = 500, overlap: int = 50):
    """Split content into overlapping chunks for better retrieval."""
    tokens = self.tokenizer.encode(content)
    chunks = []
    
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        chunk_text = self.tokenizer.decode(chunk_tokens)
        chunks.append(chunk_text)
        
        if end == len(tokens):
            break
        start = end - overlap
    
    return chunks

Why Chunking Matters:

Improved Precision: Smaller chunks mean more precise matches
Context Preservation: Overlap ensures important context isn’t lost
Efficient Processing: Manageable sizes for embedding models

3. Embedding Generation

We use Sentence Transformers for creating semantic embeddings:

self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')

# Combine title and content for richer context
text_to_embed = f"Title: {post.title}\n\nContent: {chunk}"
embeddings = self.embedding_model.encode(all_chunks)

The all-MiniLM-L6-v2 model provides excellent semantic understanding while being computationally efficient.

4. FAISS Vector Database

FAISS (Facebook AI Similarity Search) provides lightning-fast similarity search:

# Create index optimized for cosine similarity
dimension = embeddings.shape[1]
self.faiss_index = faiss.IndexFlatIP(dimension)

# Normalize for cosine similarity
faiss.normalize_L2(embeddings)
self.faiss_index.add(embeddings.astype('float32'))

5. AWS Bedrock Integration

The system uses Claude 3.5 Sonnet via Bedrock for intelligent responses:

def query_bedrock_claude(self, prompt: str, context: str) -> str:
    system_prompt = f"""You are an AI assistant that answers questions based ONLY on the provided blog post content. 

Rules:
1. ONLY use information from the provided blog post excerpts
2. If the blog posts don't contain enough information, say so explicitly
3. Always cite which blog post(s) you're referencing by title
4. Do not use any external knowledge beyond what's in the blog posts
5. Be concise but thorough in your responses

Blog post content:
{context}"""

    response = self.bedrock_client.invoke_model(
        modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
        body=json.dumps({
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": 1000,
            "system": system_prompt,
            "messages": [{"role": "user", "content": f"Question: {prompt}"}]
        })
    )
    
    return json.loads(response['body'].read())['content'][0]['text']

Installation and Setup

1. Install Dependencies

pip install boto3 sentence-transformers faiss-cpu python-frontmatter tiktoken numpy

2. Configure AWS Credentials

Choose your preferred method:

# Option A: AWS CLI
aws configure

# Option B: Environment Variables
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key

# Option C: Use IAM roles (if running on EC2)

3. Enable Bedrock Model Access

Navigate to AWS Bedrock console in your region (ap-south-1)
Go to “Model access” in the left sidebar
Request access to Claude models if not already enabled
Wait for approval (usually instant for Claude models)

4. Update Configuration

Modify the blog path in the script to match your Jekyll site:

rag = JekyllBlogRAG(
    blog_path="/your/jekyll/blog/path",
    aws_region="ap-south-1"
)

Usage Examples

Once set up, you can ask sophisticated questions:

Technical Implementation Questions:

Q: "How do I implement authentication in React applications?"
A: Based on your blog post "React Authentication Best Practices", you can implement authentication using JWT tokens with...

Architecture and Design:

Q: "What are the pros and cons of microservices architecture?"
A: In your post "Microservices vs Monoliths: A Practical Guide", you outlined several key advantages of microservices...

Performance Optimization:

Q: "How can I optimize database queries for better performance?"
A: Your blog post "Database Optimization Techniques" covers several strategies including indexing, query optimization...

Performance Considerations

Indexing Performance

Initial Setup: ~2-5 minutes for 100 blog posts
Query Time: ~100-200ms per question
Memory Usage: ~50MB for 1000 post chunks

Optimization Tips

Chunk Size Tuning: Experiment with chunk sizes (300-800 tokens)
Embedding Model Selection: Balance accuracy vs speed
Index Persistence: Save/load indices to avoid re-indexing
Batch Processing: Process multiple queries together

Advanced Features You Can Add

1. Incremental Updates

Track file modification times to only re-index changed posts:

def needs_reindexing(self, file_path: Path) -> bool:
    """Check if file needs re-indexing based on modification time."""
    last_indexed = self.get_last_indexed_time(file_path)
    file_mtime = file_path.stat().st_mtime
    return file_mtime > last_indexed

2. Semantic Filtering

Add filters for categories, tags, or date ranges:

def search_with_filters(self, query: str, categories: List[str] = None, 
                       date_range: tuple = None) -> List[Dict]:
    """Search with additional filtering criteria."""
    results = self.search_similar_chunks(query, top_k=20)
    
    if categories:
        results = [r for r in results if any(cat in r['categories'] for cat in categories)]
    
    if date_range:
        results = [r for r in results if date_range[0] <= r['date'] <= date_range[1]]
    
    return results[:5]  # Return top 5 after filtering

Extend to handle images and code snippets:

def extract_code_blocks(self, content: str) -> List[Dict]:
    """Extract and index code blocks separately."""
    code_pattern = r'```(\w+)?\n(.*?)\n```'
    matches = re.findall(code_pattern, content, re.DOTALL)
    
    return [{'language': lang, 'code': code} for lang, code in matches]

Troubleshooting Common Issues

1. No Blog Posts Found

# Debug post discovery
post_files = self.find_posts()
print(f"Checking paths: {[str(p) for p in post_patterns]}")
print(f"Found files: {[str(f) for f in post_files]}")

2. AWS Bedrock Access Errors

Verify region configuration matches your model access
Check IAM permissions for Bedrock actions
Confirm model access is granted in Bedrock console

3. Memory Issues with Large Blogs

Reduce chunk size or increase overlap
Process posts in batches
Use FAISS indices with compression

Security Considerations

1. Content Filtering

Ensure sensitive information isn’t indexed:

def should_index_post(self, post: BlogPost) -> bool:
    """Check if post should be indexed (skip drafts, private posts)."""
    if 'draft' in post.categories or 'private' in post.tags:
        return False
    return True

2. Query Sanitization

Validate and sanitize user inputs:

def sanitize_query(self, query: str) -> str:
    """Sanitize user query to prevent injection attacks."""
    # Remove potentially harmful characters
    return re.sub(r'[<>\"\'%;()&+]', '', query).strip()

3. Rate Limiting

Implement rate limiting for production use:

from time import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int = 10, time_window: int = 60):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = defaultdict(list)
    
    def allow_request(self, client_id: str) -> bool:
        now = time()
        client_requests = self.requests[client_id]
        
        # Remove old requests
        client_requests[:] = [req_time for req_time in client_requests 
                             if now - req_time < self.time_window]
        
        if len(client_requests) < self.max_requests:
            client_requests.append(now)
            return True
        
        return False

Cost Optimization

AWS Bedrock Costs

Claude 3.5 Sonnet: ~$0.003 per 1K input tokens, ~$0.015 per 1K output tokens
Typical Query: ~$0.01-0.05 per question
Monthly Budget: ~$10-50 for moderate usage

Cost Reduction Strategies

Caching: Store responses for common queries
Batch Processing: Group similar questions
Model Selection: Use smaller models for simpler queries
Context Optimization: Send only relevant chunks

Future Enhancements

1. Web Interface

Build a React/Vue frontend:

// Example API integration
const askQuestion = async (question) => {
  const response = await fetch('/api/ask', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({question})
  });
  return response.json();
};

2. Analytics Dashboard

Track popular queries and content performance:

def log_query_analytics(self, query: str, results: List[Dict], 
                       response_time: float):
    """Log query analytics for insights."""
    analytics_data = {
        'timestamp': datetime.now(),
        'query': query,
        'num_results': len(results),
        'response_time': response_time,
        'top_posts': [r['title'] for r in results[:3]]
    }
    
    # Store in database or analytics service
    self.store_analytics(analytics_data)

3. Multi-Language Support

Extend to support multiple languages:

def detect_language(self, content: str) -> str:
    """Detect content language for appropriate processing."""
    from langdetect import detect
    return detect(content)

def get_embedding_model(self, language: str) -> SentenceTransformer:
    """Select appropriate embedding model for language."""
    models = {
        'en': 'all-MiniLM-L6-v2',
        'hi': 'sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2',
        # Add more language models as needed
    }
    return SentenceTransformer(models.get(language, models['en']))

Conclusion

Building a RAG system for your Jekyll blog creates a powerful way to make your content more accessible and useful. The combination of semantic search with Claude’s reasoning capabilities provides responses that are both accurate and contextually rich.

The system we’ve built offers:

Intelligent Search: Goes beyond keyword matching
Authoritative Responses: Always grounded in your actual content
Scalable Architecture: Handles growing blog content efficiently
Cost-Effective Operation: Reasonable costs for individual blogs

Whether you’re building this for personal use or considering it for a larger content site, the principles and code provided give you a solid foundation to build upon.

Resources and Next Steps

Rofi Process Killer

2025-11-01T00:00:00+05:30

Disclaimer: This blog article has been generated with the assistance of AI to help announce and explain the features of Rofi Process Killer. While the content is AI-generated, the software itself and the ideas behind it are the result of real development work and genuine user needs.

Introducing Rofi Process Killer: A Better Way to Manage Linux Processes

Ever found yourself hunting through htop or typing ps aux | grep just to kill a misbehaving process? If you’re a Linux user who loves efficient workflows, I’ve got something that might change how you handle process management forever.

Today, I’m excited to announce Rofi Process Killer – a sleek, keyboard-driven process manager that integrates seamlessly with the popular rofi application launcher.

What is Rofi Process Killer?

Rofi Process Killer is a custom rofi module that transforms process management from a command-line chore into an intuitive, visual experience. With just a few keystrokes, you can:

View all running processes with complete command lines, CPU usage, and memory consumption
Kill processes instantly with intelligent graceful termination (SIGTERM) followed by force kill (SIGKILL) if needed
See full command arguments – no more truncated process names
Filter processes in real-time using rofi’s built-in search capabilities

Why I Built This

As someone who frequently works with multiple development environments, Docker containers, and various system processes, I was constantly frustrated by the traditional process management workflow:

Open terminal
Run ps aux or htop
Search for the problematic process
Note down the PID
Kill it with kill command

This felt clunky, especially when using tiling window managers where quick, keyboard-driven actions are paramount. I wanted something that felt as smooth as launching applications with rofi – and that’s exactly what this tool delivers.

Key Features

Complete Process Information

Unlike basic process viewers, Rofi Process Killer shows you everything you need:

Process ID (PID)
CPU usage percentage
Memory usage percentage
Process owner
Full command line with all arguments (no truncation!)

Smart Text Wrapping

Long command lines don’t get cut off – they wrap intelligently across multiple lines, so you can see exactly what each process is doing.

Safe Process Termination

The tool follows best practices for process management:

First attempts graceful termination (SIGTERM)
If the process doesn’t respond, automatically escalates to force kill (SIGKILL)
Provides desktop notifications for feedback

Beautiful Dark Theme

Includes a custom rofi theme optimized for process management with:

Clean, readable monospace font
Dark color scheme that’s easy on the eyes
Proper spacing for wrapped text
Visual distinction between normal and selected processes

Installation

For Arch Linux Users

# Install from AUR 
yay -S rofi-process-killer

# Or install manually from GitHub
git clone https://github.com/madhur/rofi-process-killer.git
cd rofi-process-killer
chmod +x rofi-process-killer.sh
cp rofi-process-killer.sh ~/.local/bin/
cp process-killer.rasi ~/.config/rofi/themes/

Usage

Launch with a simple command:

rofi -modi "processes:rofi-process-killer" -show processes -theme ~/.config/rofi/themes/process-killer.rasi

Or create a keyboard shortcut in your window manager:

# i3/sway example
bindsym $mod+Shift+p exec rofi -modi "processes:rofi-process-killer" -show processes

Perfect for Power Users

This tool shines in environments where efficiency matters:

Tiling Window Managers: i3, sway, bspwm, etc.
Development Workflows: Quickly kill runaway dev servers, Docker containers, or build processes
System Administration: Fast process management without leaving your current workspace
General Productivity: Any scenario where you need quick access to process management

Technical Implementation

For those curious about the internals, the tool is built as a rofi script mode that:

Uses ps auxww with unlimited width to capture complete command lines
Implements intelligent process filtering to exclude kernel threads
Provides error handling for permission-denied scenarios
Integrates with the desktop notification system for user feedback

The custom rofi theme leverages advanced CSS-like styling to enable text wrapping and optimize the visual layout for process information.

What’s Next?

I’m already working on several enhancements:

Process filtering options (by user, CPU usage, memory usage)
Process tree view to show parent-child relationships
Resource usage graphs for visual process monitoring
Integration with systemd for service management
Configuration file for customizable display options

Try It Today

Ready to streamline your process management workflow?

🔗 GitHub Repository: https://github.com/madhur/rofi-process-killer

The repository includes:

Complete installation instructions
Usage examples and screenshots
Custom theme files
Keyboard shortcut configurations for popular window managers

Contributing

This is an open-source project, and I welcome contributions! Whether you want to:

Report bugs or suggest features
Improve the documentation
Add new functionality
Create additional themes

Feel free to open issues or submit pull requests on GitHub.

Final Thoughts

Rofi Process Killer represents what I love about the Linux ecosystem – the ability to create tools that perfectly fit your workflow. It’s fast, efficient, and keyboard-driven, embodying the Unix philosophy of doing one thing well.

If you’re someone who appreciates well-crafted developer tools and efficient workflows, give it a try. I think you’ll find it becomes an indispensable part of your daily Linux experience.

Have you tried Rofi Process Killer? I’d love to hear your feedback! Connect with me on GitHub or share your experience in the comments below.

Tags: #Linux #Rofi #ProcessManagement #OpenSource #ProductivityTools #TilingWindowManagers #Developer Tools

Configuring AWS ALB Timeouts

2025-10-01T00:00:00+05:30

Author’s Note: I recently encountered this exact library issue. After solving it, I used Claude AI to help me structure this troubleshooting guide based on my experience. The problem and solution are genuine - I just wanted to create a quick resource that others could benefit from.

ALB Idle Timeout Configuration: Best Practices to Prevent 5xx Errors

If you’ve ever encountered mysterious 502 Bad Gateway or 504 Gateway Timeout errors from your AWS Application Load Balancer (ALB), you’re not alone. Many of these issues stem from improperly configured timeout settings between your ALB and backend applications. In this article, we’ll explore the best practices for configuring ALB idle timeouts.

Understanding the Problem

AWS Application Load Balancers manage connections between clients and your backend applications. When timeout configurations are misaligned, you create race conditions that result in intermittent 5xx errors and degraded user experience.

The root cause typically involves two types of timeouts that must work in harmony:

Request timeouts: How long to wait for a single request/response cycle
Keep-alive (idle) timeouts: How long to maintain connections between requests

Request Timeout Best Practices

The Golden Rule: Application Timeout < ALB Timeout

Your application’s request timeout should always be shorter than your ALB’s timeout. Here’s the recommended hierarchy:

ALB Request Timeout: 60 seconds (default)
Application Timeout: 50-55 seconds
Database/External API Timeout: 45 seconds or less

Why This Matters

When your application timeout equals or exceeds the ALB timeout, you create a dangerous race condition:

Request arrives at ALB and forwards to your application
Both timers start counting down simultaneously
Application timeout hits first and closes the connection
ALB is still waiting for a response when connection terminates
ALB generates 5xx error instead of receiving your application’s graceful error response

Keep-Alive Timeout Best Practices

The Reverse Rule: Application Keep-Alive > ALB Idle Timeout

Unlike request timeouts, keep-alive timeouts follow the opposite pattern:

ALB Idle Timeout: 60 seconds (default)
Application Keep-Alive: 65+ seconds
Database Connection Pool Idle: 70+ seconds

Why Application Keep-Alive Should Be Higher

If your application’s keep-alive timeout is shorter than ALB’s idle timeout:

Client completes request through ALB to application
Application closes connection after its shorter keep-alive period
ALB still considers connection active and attempts reuse
Next request fails with 502 Bad Gateway because connection is closed

Advanced Configuration Scenarios

High-Traffic Applications

For applications handling high request volumes:

ALB Idle Timeout: 30 seconds (reduced)
Application Keep-Alive: 35 seconds
Request Timeout: 25 seconds

Shorter timeouts free up connections faster and improve resource utilization.

Long-Running Operations

For applications with legitimate long-running operations:

ALB Request Timeout: 300 seconds (5 minutes)
Application Timeout: 280 seconds
Keep-Alive: 310 seconds

Consider moving truly long operations to asynchronous processing with status polling endpoints.

Microservices Architecture

For service-to-service communication:

External ALB (Internet-facing):
  - Request Timeout: 60s
  - Idle Timeout: 60s

Internal ALB (Service-to-service):
  - Request Timeout: 30s
  - Idle Timeout: 30s

Application Configuration:
  - Request Timeout: 25s
  - Keep-Alive: 35s

Monitoring and Troubleshooting

Key Metrics to Monitor

ALB CloudWatch Metrics:

HTTPCode_ELB_5XX_Count: Track 5xx errors from ALB
TargetResponseTime: Monitor backend response times
ActiveConnectionCount: Watch connection patterns

Application Metrics:

Connection pool utilization
Request timeout occurrences
Keep-alive connection reuse rates

Common Error Patterns

502 Bad Gateway:

Often indicates application closed connection unexpectedly
Check: Application keep-alive < ALB idle timeout
Check: Application crashed or became unresponsive

504 Gateway Timeout:

ALB timeout expired waiting for response
Check: Application timeout ≥ ALB request timeout
Check: Long-running operations without proper timeout handling

Debugging Commands

Test connection behavior:

# Test keep-alive behavior
curl -H "Connection: keep-alive" -v http://your-alb-endpoint/

# Monitor connection reuse
curl -H "Connection: keep-alive" -v \
  http://your-alb-endpoint/endpoint1 \
  http://your-alb-endpoint/endpoint2

Check ALB configuration:

aws elbv2 describe-load-balancer-attributes \
  --load-balancer-arn arn:aws:elasticloadbalancing:...

Best Practices Summary

Request Timeouts: Application timeout should be 10-15% shorter than ALB timeout
Keep-Alive Timeouts: Application keep-alive should be 5-10 seconds longer than ALB idle timeout
Monitor Religiously: Set up alerts for 5xx error rates and connection metrics
Test Thoroughly: Use load testing tools to validate timeout behavior under various conditions
Document Configuration: Maintain clear documentation of timeout values across your infrastructure
Gradual Changes: Modify timeout values incrementally and monitor impact

Conclusion

Proper timeout configuration is crucial for maintaining reliable web applications behind AWS Application Load Balancers. By following the principles outlined in this guide—keeping application request timeouts shorter than ALB timeouts and application keep-alive timeouts longer than ALB idle timeouts—you can eliminate most timeout-related 5xx errors.

Remember that timeout configuration is not a one-size-fits-all solution. Consider your application’s specific requirements, traffic patterns, and operational constraints when implementing these best practices. Regular monitoring and testing will help you fine-tune these settings for optimal performance and reliability.

The investment in properly configuring these timeouts will pay dividends in improved user experience, reduced operational overhead, and fewer 3 AM troubleshooting sessions.

Non Blocking I/O

2025-09-01T00:00:00+05:30

When I first learned about non-blocking I/O, I had a burning question: how does a process know when I/O operations complete? Is there some background kernel process constantly checking for data and then notifying user-space applications? The answer reveals one of the most elegant solutions in systems programming.

The Problem with Blocking I/O

To understand non-blocking I/O, let’s first examine what happens with traditional blocking I/O. When your program makes a system call like read() or recv(), the calling thread is suspended until the operation completes. For a web server handling thousands of connections, this means you’d need thousands of threads – each consuming memory and creating context-switching overhead.

// Blocking I/O - thread waits here until data arrives
int bytes = recv(socket_fd, buffer, sizeof(buffer), 0);

This approach doesn’t scale well. A server with 10,000 connections would need 10,000 threads, consuming gigabytes of memory just for thread stacks.

How Non-Blocking I/O Works

Non-blocking I/O flips this model on its head. Instead of waiting for I/O to complete, operations return immediately with either data (if available) or an indication that no data is ready.

// Non-blocking I/O - returns immediately
int flags = fcntl(socket_fd, F_GETFL, 0);
fcntl(socket_fd, F_SETFL, flags | O_NONBLOCK);

int bytes = recv(socket_fd, buffer, sizeof(buffer), 0);
if (bytes == -1 && errno == EAGAIN) {
    // No data available right now, try again later
}

But this raises the obvious question: how do you know when to “try again later”? The naive approach would be to continuously poll all file descriptors in a loop, but that would waste enormous amounts of CPU cycles.

The Event Loop: Efficient I/O Multiplexing

Modern operating systems provide sophisticated APIs to efficiently monitor multiple file descriptors. This is where the “event loop” comes in – but it’s far more elegant than a simple while loop.

Linux: epoll

Linux provides epoll, which allows you to monitor hundreds or thousands of file descriptors efficiently:

// Create an epoll instance
int epoll_fd = epoll_create1(0);

// Add a socket to the interest list
struct epoll_event event;
event.events = EPOLLIN;  // Monitor for readable data
event.data.fd = socket_fd;
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, socket_fd, &event);

// Wait for events (this blocks until data is available)
struct epoll_event events[MAX_EVENTS];
int ready_fds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);

for (int i = 0; i < ready_fds; i++) {
    // Process ready file descriptor
    handle_ready_socket(events[i].data.fd);
}

The Key Optimization

Here’s the crucial insight: epoll_wait() blocks until at least one file descriptor has data available. This means:

No busy waiting: The process sleeps when there’s no work to do
Efficient scaling: Monitoring 1,000 FDs is nearly as fast as monitoring 10 FDs
Event-driven: You only process FDs that actually have data ready

The computational complexity is O(number of ready events), not O(total FDs monitored). You could monitor 100,000 connections, but if only 10 have data ready, you only process 10 events.

Cross-Platform Event Mechanisms

Different operating systems provide similar but distinct APIs:

Linux: epoll (edge-triggered and level-triggered modes)
BSD/macOS: kqueue (supports files, network, timers, and more)
Windows: I/O Completion Ports (IOCP) with a different asynchronous model
Portable: select and poll (available everywhere but less efficient)

Real-World Example: A Simple HTTP Server

Here’s how a basic non-blocking HTTP server might work:

int server_fd = socket(AF_INET, SOCK_STREAM, 0);
// Set up server socket and bind to port...
fcntl(server_fd, F_SETFL, O_NONBLOCK);
listen(server_fd, SOMAXCONN);

int epoll_fd = epoll_create1(0);
// Add server socket to epoll...

while (1) {
    int ready = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
    
    for (int i = 0; i < ready; i++) {
        if (events[i].data.fd == server_fd) {
            // New connection
            int client_fd = accept(server_fd, NULL, NULL);
            fcntl(client_fd, F_SETFL, O_NONBLOCK);
            // Add client to epoll...
        } else {
            // Data ready on existing connection
            handle_client_request(events[i].data.fd);
        }
    }
}

This single-threaded server can handle thousands of concurrent connections efficiently.

Answering the Original Question

So, to answer the original question: there’s no background kernel process constantly checking for data. Instead:

The kernel maintains readiness state for each file descriptor
Your application explicitly asks “which FDs are ready?” using epoll_wait() or similar
The kernel efficiently returns only the FDs that have data available
Your application processes those ready FDs and then asks again

It’s a pull model, not a push model. The application pulls information about ready file descriptors when it’s ready to process them.

Modern Frameworks and Languages

This pattern is so fundamental that it’s built into many high-level frameworks:

Node.js: Uses libuv, which wraps epoll/kqueue/IOCP
Go: Goroutines use an internal poller for network operations
Rust: Tokio provides async I/O built on these primitives
Java: NIO uses select/epoll under the hood
Python: asyncio leverages these same mechanisms

Understanding non-blocking I/O helps you write more efficient programs and better understand how modern high-performance applications work under the hood.

Key Takeaways

Non-blocking I/O doesn’t use callbacks at the kernel level – it uses efficient polling mechanisms
Event loops aren’t constantly spinning – they block until events are ready
Modern OS APIs like epoll scale to thousands of connections with minimal overhead
Single-threaded event loops can outperform multi-threaded blocking approaches for I/O-bound workloads
The pattern is ubiquitous in modern programming languages and frameworks

The elegance of non-blocking I/O lies in its simplicity: instead of creating threads for each connection, you let the kernel tell you when work is ready to be done.

References:

Troubleshooting Shared Library Issues on Linux

2025-08-01T00:00:00+05:30

Author’s Note: I recently encountered this exact library issue with qimgv and OpenCV on my Arch Linux system. After solving it, I used Claude AI to help me structure this troubleshooting guide based on my experience. The problem and solution are genuine - I just wanted to create a comprehensive resource that others could benefit from.

If you’ve been using Linux for a while, you’ve probably encountered the dreaded “cannot open shared object file” error at least once. This error can be frustrating, especially when you’re trying to run a perfectly good application that worked yesterday. In this guide, we’ll explore what causes these errors and how to fix them systematically.

Understanding the Error

When you see an error like this:

qimgv: error while loading shared libraries: libopencv_imgproc.so.411: cannot open shared object file: No such file or directory

It means that the application (in this case qimgv) is looking for a specific shared library file that it can’t find on your system. Shared libraries are collections of code that multiple programs can use simultaneously, making your system more efficient.

Common Causes

Several scenarios can lead to this error:

Missing library: The required library isn’t installed on your system
Version mismatch: You have a different version of the library than what the application expects
Incorrect library path: The library exists but isn’t in the system’s search path
Corrupted installation: The library files are damaged or incomplete

Step-by-Step Troubleshooting

Step 1: Check What Libraries Are Available

First, let’s see what versions of the library you have installed:

ldconfig -p | grep [library_name]

For our OpenCV example:

ldconfig -p | grep opencv

This command will show you all the OpenCV libraries installed on your system along with their versions and paths.

Step 2: Identify the Version Mismatch

In our case, the output showed:

libopencv_imgproc.so.412 (libc6,x86-64) => /usr/lib/libopencv_imgproc.so.412

The application was looking for libopencv_imgproc.so.411 but we had version 4.12 (.412) installed. This is a classic version mismatch.

Step 3: Check What the Application Actually Needs

To see exactly what libraries an application is trying to link to:

ldd $(which application_name) | grep [library_name]

For example:

ldd $(which qimgv) | grep opencv

This will show you all the OpenCV libraries that qimgv expects to find.

Step 4: Choose Your Fix Strategy

You have several options depending on the situation:

Option A: Install the Correct Version (Recommended for missing libraries)

If the library is completely missing, install it using pacman:

sudo pacman -S opencv

You can also search for OpenCV-related packages:

pacman -Ss opencv

If you need development headers and additional libraries:

sudo pacman -S opencv vtk hdf5

Option B: Create Symbolic Links (Best for version mismatches)

When you have a newer version of the library that’s compatible with the older version the application expects, create symbolic links:

sudo ln -s /usr/lib/libopencv_imgproc.so.412 /usr/lib/libopencv_imgproc.so.411

You might need to create links for multiple libraries. Based on the ldd output, you might need:

sudo ln -s /usr/lib/libopencv_core.so.412 /usr/lib/libopencv_core.so.411
sudo ln -s /usr/lib/libopencv_imgcodecs.so.412 /usr/lib/libopencv_imgcodecs.so.411
sudo ln -s /usr/lib/libopencv_highgui.so.412 /usr/lib/libopencv_highgui.so.411

After creating symbolic links, update the library cache:

sudo ldconfig

Option C: Reinstall the Application

Sometimes the easiest solution is to reinstall the problematic application to ensure all dependencies are properly resolved:

sudo pacman -R qimgv
sudo pacman -S qimgv

You can also use pacman to check what packages depend on a specific library:

pacman -Qi opencv

Or find which package provides a specific file:

pacman -Qo /usr/lib/libopencv_imgproc.so.412

Step 5: Verify the Fix

After applying your chosen solution, test the application:

qimgv

If it still doesn’t work, run the ldd command again to see if there are other missing libraries.

Advanced Troubleshooting

Checking Library Paths

If the library exists but isn’t being found, check your library search paths:

echo $LD_LIBRARY_PATH

You can temporarily add a directory to the search path:

export LD_LIBRARY_PATH=/path/to/library:$LD_LIBRARY_PATH

Finding Libraries by Name

To find all instances of a library on your system:

find /usr -name "libopencv_imgproc.so*" 2>/dev/null

You can also use pacman to find which package provides a file:

pacman -F libopencv_imgproc.so

Or search for packages containing specific files:

pacman -Fx opencv

### Checking Library Dependencies

To see what other libraries a specific library depends on:

```bash
ldd /usr/lib/libopencv_imgproc.so.412

Prevention Tips

Use pacman: Install software through pacman when possible, as it handles dependencies automatically.
Keep your system updated: Regular updates often resolve compatibility issues:
```
sudo pacman -Syu
```
Use AUR helpers carefully: When installing from AUR, be aware that it might create dependency conflicts. Consider using yay or paru for better dependency management:
```
yay -S package-name
```
Check for orphaned packages: Clean up unused dependencies:
```
sudo pacman -Rns $(pacman -Qtdq)
```
Use containerization: For complex applications with specific library requirements, consider using Docker or similar containerization tools.

When Symbolic Links Might Not Work

While symbolic links are often a quick fix, they’re not always appropriate:

Major version differences: Linking between major versions (e.g., OpenCV 3.x to 4.x) can cause crashes
ABI incompatibility: Different library versions might have incompatible application binary interfaces
Security concerns: Older applications might have security vulnerabilities when used with newer libraries

In these cases, consider using the application’s official repositories, building from source, or using containerization.

Conclusion

Shared library errors are common in Linux but usually straightforward to fix once you understand the underlying cause. The key is to systematically diagnose the problem: check what you have, identify what’s missing or mismatched, and choose the appropriate fix strategy.

Remember that while symbolic links are often the quickest solution for version mismatches, they’re not always the safest long-term approach. When possible, prefer proper package management and keep your system updated to avoid these issues altogether.

The next time you encounter a “cannot open shared object file” error, you’ll have the tools and knowledge to diagnose and fix it efficiently. Happy troubleshooting!

Row level locking in database

2025-07-09T00:00:00+05:30

Quick refereshers on how locks are implemented in databases.

Exclusive Locks - The row is locked for read and update. Until the exclusive lock is released, no reads and writes are allowed. Only one transaction can acquire an exclusive lock.
Shared Lock - The row is locked for update. However, readers are allowed to read. Multiple transactions are allowed to acquire shared locks.

Major points

If the transaction has shared lock on resource, other transaction cannot acquire exclusive lock on it.
A process cannot acquire a shared lock on a resource that is locked by an exclusive lock.

Optimistic vs Pessimistic Locking

Optimistic - Let multiple transactions be allowed to update the row concurrently. One one will succeed and others will fail.
Pessimistic - Only one transaction is allowed to update the row. Others are put to wait.

Transaction Isolation

What happens when a transaction tries to read a row updated by another transaction?

Problems when transaction isolation is not done

Dirty Read — Let’s take a situation where one transaction updates a row or a table but does not commit the changes. If the database lets another transaction read those changes (before it’s committed) then it’s called a dirty read. Why? Let’s say the first transaction rolls back its changes. The other transaction which read the row/table has stale data.
Non-repeatable read — Another side effect of concurrent execution of transactions is that consecutive reads can retrieve different results if you allow another transaction to do updates in between
Phantom read — In a similar situation as above, if one transaction does two reads of the same query, but another transaction inserts or deletes new rows leading to a change in the number of rows retrieved by the first transaction in its second read, it’s called a Phantom read.

Levels of Isolation

Read uncomitted - This level of isolation lets other transactions read data that was not committed to the database by other transactions.
Read committed — This, as the name suggests, lets other transactions only read data that is committed to the database. While this looks like an ideal level of isolation, it only solves the dirty read problem mentioned above. If a transaction is updating a row, and another transaction tries to access it, it won’t be able to. But this can still cause non-repeatable and phantom reads because this applies only to updates and not read queries.
Repeatable read - To counter the transactions from getting inconsistent data, we need a higher level of isolation and that is offered by repeatable read. In this, the resource is locked throughout the transaction. So, if the transaction contains two select queries and in between, if another transaction tries to update the same rows, it would be blocked from doing so. This isolation level is not immune to phantom reads though it helps against non-repeatable reads. This is the default level of isolation in many databases.
Serializable — This is the highest level of isolation. In this, all concurrent transactions “appear” to be executed serially. Pay attention to how I said it appears to be. That’s because it’s not truly serially or sequentially executed. This level works against phantom reads as well.

References

https://medium.com/inspiredbrilliance/what-are-database-locks-1aff9117c290

Run elasticsearch in docker

2025-06-01T00:00:00+05:30

If you’re running Elasticsearch in Docker locally, you might encounter a frustrating issue where your cluster suddenly goes into a red (unhealthy) state, even though everything seems fine. This often happens when your local disk usage exceeds 90%.

I tend to mount the elasticsearch data directory into my host file system to preserve data.

version: "3.9"
services:
  elasticsearch:
    image: elasticsearch:9.0.2
    environment:
      - discovery.type=single-node
      - ES_JAVA_OPTS=-Xms1g -Xmx1g
      - xpack.security.enabled=false
    volumes:
      - ./es_data:/usr/share/elasticsearch/data
    ports:
      - target: 9200
        published: 9200
    networks:
      - elastic

  kibana:
    image: kibana:9.0.2
    ports:
      - target: 5601
        published: 5601
    depends_on:
      - elasticsearch
    networks:
      - elastic      

volumes:
  es_data:
    driver: local

networks:
  elastic:
    name: elastic
    driver: bridge

The Problem: Disk Space Watermarks

When your local filesystem usage exceeds 90%, Elasticsearch triggers its high watermark threshold, causing the cluster to enter a red state. This is particularly common on development machines where disk space is often limited.

Solution: Adjusting Watermark Thresholds

If you are facing the same issue, to resolve this, run this curl command which adjusts the disk space percentage at which elasticsearch goes into unhealthy state.

curl -X PUT "localhost:9200/_cluster/settings" \
  -H 'Content-Type: application/json' \
  -d '{
    "persistent": {
      "cluster.routing.allocation.disk.watermark.low": "95%",
      "cluster.routing.allocation.disk.watermark.high": "97%",
      "cluster.routing.allocation.disk.watermark.flood_stage": "98%"
    }
  }'

Check your cluster health:

curl "localhost:9200/_cluster/health?pretty"

HTTP Certificate Pinning

2025-05-01T00:00:00+05:30

Certificate pinning code for Java

package org.example;

import javax.net.ssl.*;
import java.io.*;
import java.net.URL;
import java.security.MessageDigest;
import java.security.cert.Certificate;
import java.security.cert.X509Certificate;
import java.util.Base64;

public class CertificatePinningClient {

    // SHA-256 fingerprint of the expected certificate
    // You need to replace this with the actual certificate fingerprint of madhur.co.in
    private static final String EXPECTED_CERT_SHA256 = "YOUR_CERT_FINGERPRINT_HERE";

    public static void main(String[] args) {
        try {
            String fingerprint = getCertificateFingerprint("https://localhost:8443");
            System.out.println(fingerprint);
          //  String response = makeSecureRequest("https://madhur.co.in");
           // System.out.println("Response: " + response);
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
            e.printStackTrace();
        }
    }

    public static String makeSecureRequest(String urlString) throws Exception {
        URL url = new URL(urlString);
        HttpsURLConnection connection = (HttpsURLConnection) url.openConnection();

        // Set up custom SSL context with certificate pinning
        SSLContext sslContext = SSLContext.getInstance("TLS");
        sslContext.init(null, new TrustManager[]{new PinningTrustManager()}, null);
        connection.setSSLSocketFactory(sslContext.getSocketFactory());

        // Set request properties
        connection.setRequestMethod("GET");
        connection.setConnectTimeout(10000);
        connection.setReadTimeout(10000);

        // Make the request
        connection.getResponseCode();

        // Read the response
        StringBuilder response = new StringBuilder();
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(connection.getInputStream()))) {
            String line;
            while ((line = reader.readLine()) != null) {
                response.append(line).append("\n");
            }
        }

        connection.disconnect();
        return response.toString();
    }

    // Custom TrustManager that implements certificate pinning
    private static class PinningTrustManager implements X509TrustManager {

        @Override
        public void checkClientTrusted(X509Certificate[] chain, String authType) {
            // Not used for client certificates in this example
        }

        @Override
        public void checkServerTrusted(X509Certificate[] chain, String authType)
                throws java.security.cert.CertificateException {

            if (chain == null || chain.length == 0) {
                throw new java.security.cert.CertificateException("Certificate chain is empty");
            }

            // Get the server certificate (first in chain)
            X509Certificate serverCert = chain[0];

            try {
                // Calculate SHA-256 fingerprint of the certificate
                MessageDigest md = MessageDigest.getInstance("SHA-256");
                byte[] certBytes = serverCert.getEncoded();
                byte[] fingerprint = md.digest(certBytes);
                String certFingerprint = Base64.getEncoder().encodeToString(fingerprint);

                System.out.println("Server certificate fingerprint: " + certFingerprint);
                System.out.println("Expected fingerprint: " + EXPECTED_CERT_SHA256);

                // Check if the certificate matches our pinned certificate
                if (!EXPECTED_CERT_SHA256.equals(certFingerprint)) {
                    throw new java.security.cert.CertificateException("Certificate pinning failed! " +
                            "Expected: " + EXPECTED_CERT_SHA256 +
                            ", Got: " + certFingerprint);
                }

                System.out.println("Certificate pinning validation passed!");

            } catch (Exception e) {
                throw new java.security.cert.CertificateException("Error validating certificate: " + e.getMessage(), e);
            }
        }

        @Override
        public X509Certificate[] getAcceptedIssuers() {
            return new X509Certificate[0];
        }
    }

    // Utility method to get certificate fingerprint for a given URL
    // Run this first to get the actual certificate fingerprint
    public static String getCertificateFingerprint(String urlString) throws Exception {
        URL url = new URL(urlString);
        HttpsURLConnection connection = (HttpsURLConnection) url.openConnection();
        connection.connect();

        Certificate[] certificates = connection.getServerCertificates();
        if (certificates.length > 0) {
            X509Certificate cert = (X509Certificate) certificates[0];
            MessageDigest md = MessageDigest.getInstance("SHA-256");
            byte[] fingerprint = md.digest(cert.getEncoded());
            return Base64.getEncoder().encodeToString(fingerprint);
        }

        connection.disconnect();
        throw new Exception("No certificates found");
    }


}

Combine videos using ffmpeg

2025-04-01T00:00:00+05:30

I wanted to combine videos using the tool. I had two videos in 9x16 format and wanted to combine side by side.

With a little bit of trial and error, I was able to achieve this using ffmpeg

The exact command is

ffmpeg -i video1.mp4 -i video2.mp4 -filter_complex "[0:v]scale=540:960:force_original_aspect_ratio=decrease,pad=540:960:(ow-iw)/2:(oh-ih)/2[v0];[1:v]scale=540:960:force_original_aspect_ratio=decrease,pad=540:960:(ow-iw)/2:(oh-ih)/2[v1];[v0][v1]hstack=inputs=2[v]" -map "[v]" -map 0:a -c:v libx264 -c:a aac output.mp4

Breaking Down the Command

Let’s understand what each part of this complex command does:

Input Files

-i video1.mp4 -i video2.mp4

This specifies our two input videos. FFmpeg will reference them as [0] and [1] respectively.

Filter Chain Explanation

The -filter_complex parameter contains several operations:

1. Scale and Pad First Video:

[0:v]scale=540:960:force_original_aspect_ratio=decrease,pad=540:960:(ow-iw)/2:(oh-ih)/2[v0]

scale=540:960: Resize to 540x960 pixels
force_original_aspect_ratio=decrease: Maintain aspect ratio, making the video smaller if needed
pad=540:960:(ow-iw)/2:(oh-ih)/2: Add padding to center the video in a 540x960 frame
[v0]: Label this processed video as “v0”

2. Scale and Pad Second Video:

[1:v]scale=540:960:force_original_aspect_ratio=decrease,pad=540:960:(ow-iw)/2:(oh-ih)/2[v1]

Same process for the second video, labeled as “v1”.

3. Horizontal Stack:

[v0][v1]hstack=inputs=2[v]

hstack: Horizontally stack the two processed videos
inputs=2: Specify we’re combining 2 video streams
[v]: Label the final combined video

Output Mapping and Encoding

-map "[v]" -map 0:a -c:v libx264 -c:a aac output.mp4

-map "[v]": Use our combined video stream
-map 0:a: Use audio from the first video
-c:v libx264: Encode video using H.264 codec
-c:a aac: Encode audio using AAC codec

Its great that in Linux even without tools such as Adobe Premiere Pro, you can achieve so much.

Case of sudden Kafka lag drop

2025-03-01T00:00:00+05:30

Recently we faced an interesting issue in our data pipeline where we observed Kafka consumer lag suddenly dropping whenever it reached a certain threshold. This behavior was unexpected and potentially concerning, as consumer lag—the difference between the latest message produced and the message currently being processed by consumers—should typically decrease gradually as consumers process messages.

The Problem

Our application was relatively simple, with two instances of pods running in Kubernetes. Since we didn’t need high throughput, the Kafka topic had only a single partition. The screenshot below shows the behavior we observed.

As visible in the graph, the consumer lag would build up gradually but then suddenly drop at certain points, creating a sawtooth pattern rather than the expected gradual decline.

Investigation

When facing such an issue with Kafka lag abruptly dropping, one of the first things to check is the retention.bytes setting of the topic and its partitions. It’s possible that when a topic reaches its retention.bytes capacity, Kafka will start deleting the oldest log segment files. The size of these segment files is defined by the segment.bytes property. To inspect our topic’s configuration, we ran:

./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic single_partition_topic --describe 

Topic: single_partition_topic	TopicId: tNA9_CAwT62MzHsW88b_Ig	PartitionCount: 1	ReplicationFactor: 2	Configs: segment.bytes=104857600,retention.bytes=173741824
	Topic: single_partition_topic	Partition: 0	Leader: 1	Replicas: 1,3	Isr: 1,3

Root Cause Analysis

The output revealed that our topic had:

A retention.bytes of 173MB
A segment.bytes of approximately 100MB

This configuration meant that for this topic, the Kafka server could maintain at most two segment log files:

The first segment file (up to 100MB)
A second segment file (which could only grow to about 73MB before hitting the total retention limit)

When the second file would reach around 73MB, Kafka would be forced to delete the first file entirely to stay within the retention limit. This deletion caused the earliest available offset in the topic to jump forward, resulting in the abrupt consumer lag drop we were observing. This problem is particularly likely to occur with topics that have a low number of partitions, since the retention.bytes setting is applied at the partition level, not at the topic level.

Solution

The solution was straightforward: increase the retention.bytes property of the topic to allow for more data retention. We executed:

./kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name single_partition_topic --alter --add-config retention.bytes=1073741824

With this command, we increased the retention.bytes property from 173MB to 1073MB (approximately 1GB), which solved our problem by allowing more headroom for message retention.

To verify that the setting was applied correctly, we ran:

./kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name single_partition_topic --describe --all

All configs for topic single_partition_topic are:
  compression.type=producer sensitive=false synonyms={DEFAULT_CONFIG:compression.type=producer}
  leader.replication.throttled.replicas= sensitive=false synonyms={}
  message.downconversion.enable=true sensitive=false synonyms={DEFAULT_CONFIG:log.message.downconversion.enable=true}
  min.insync.replicas=1 sensitive=false synonyms={DEFAULT_CONFIG:min.insync.replicas=1}
  segment.jitter.ms=0 sensitive=false synonyms={}
  cleanup.policy=delete sensitive=false synonyms={DEFAULT_CONFIG:log.cleanup.policy=delete}
  flush.ms=9223372036854775807 sensitive=false synonyms={}
  follower.replication.throttled.replicas= sensitive=false synonyms={}
  segment.bytes=104857600 sensitive=false synonyms={STATIC_BROKER_CONFIG:log.segment.bytes=104857600, DEFAULT_CONFIG:log.segment.bytes=1073741824}
  retention.ms=86400000 sensitive=false synonyms={}
  flush.messages=9223372036854775807 sensitive=false synonyms={DEFAULT_CONFIG:log.flush.interval.messages=9223372036854775807}
  message.format.version=2.8-IV1 sensitive=false synonyms={DEFAULT_CONFIG:log.message.format.version=2.8-IV1}
  max.compaction.lag.ms=9223372036854775807 sensitive=false synonyms={DEFAULT_CONFIG:log.cleaner.max.compaction.lag.ms=9223372036854775807}
  file.delete.delay.ms=60000 sensitive=false synonyms={DEFAULT_CONFIG:log.segment.delete.delay.ms=60000}
  max.message.bytes=1048588 sensitive=false synonyms={DEFAULT_CONFIG:message.max.bytes=1048588}
  min.compaction.lag.ms=0 sensitive=false synonyms={DEFAULT_CONFIG:log.cleaner.min.compaction.lag.ms=0}
  message.timestamp.type=CreateTime sensitive=false synonyms={DEFAULT_CONFIG:log.message.timestamp.type=CreateTime}
  preallocate=false sensitive=false synonyms={DEFAULT_CONFIG:log.preallocate=false}
  min.cleanable.dirty.ratio=0.5 sensitive=false synonyms={DEFAULT_CONFIG:log.cleaner.min.cleanable.ratio=0.5}
  index.interval.bytes=4096 sensitive=false synonyms={DEFAULT_CONFIG:log.index.interval.bytes=4096}
  unclean.leader.election.enable=false sensitive=false synonyms={DEFAULT_CONFIG:unclean.leader.election.enable=false}
  retention.bytes=1073741824 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.bytes=1073741824, STATIC_BROKER_CONFIG:log.retention.bytes=173741824, DEFAULT_CONFIG:log.retention.bytes=-1}
  delete.retention.ms=86400000 sensitive=false synonyms={DEFAULT_CONFIG:log.cleaner.delete.retention.ms=86400000}
  segment.ms=604800000 sensitive=false synonyms={}
  message.timestamp.difference.max.ms=9223372036854775807 sensitive=false synonyms={DEFAULT_CONFIG:log.message.timestamp.difference.max.ms=9223372036854775807}
  segment.index.bytes=10485760 sensitive=false synonyms={DEFAULT_CONFIG:log.index.size.max.bytes=10485760}

This command will show all configurations for the topic with a DEFAULT or DYNAMIC source column, where:

DYNAMIC means the configuration was explicitly set for this specific topic
DEFAULT means the configuration is inherited from the broker-level defaults

In the output (partially shown below), we could confirm the change was successful:

 retention.bytes=1073741824 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.bytes=1073741824, STATIC_BROKER_CONFIG:log.retention.bytes=173741824, DEFAULT_CONFIG:log.retention.bytes=-1}

The presence of DYNAMIC_TOPIC_CONFIG:retention.bytes=1073741824 indicates that our topic-level override was successfully applied.

Best Practices and Additional Considerations

When configuring Kafka topics, keep these points in mind:

Balance retention settings: Remember that both retention.bytes and retention.ms (time-based retention) can trigger log segment deletion. Configure both appropriately for your use case.
Monitor partition growth: For topics with few partitions, monitor data growth closely and set retention limits accordingly.
Consider partition count: If you need to retain more data and scaling consumption throughput is desirable, consider increasing the partition count.
Check broker defaults: The default retention settings are defined in the Kafka broker’s server.properties file under log.retention.bytes and log.retention.hours. Be aware of these defaults when creating new topics.

Conclusion

Sudden drops in Kafka consumer lag can often be traced back to retention settings causing older messages to be deleted. By understanding how retention.bytes and segment.bytes interact, particularly at the partition level, you can properly size these settings to match your data retention needs.

If you’re seeing unexpected behavior in your Kafka metrics, checking these configuration parameters should be one of your first troubleshooting steps.

Coding it my way

Block Storage vs Object Storage

Block Storage

How does it work?

Object Storage

Block Storage vs Object Storage

Systemd timer running on every boot

Problem Description

Environment

Observed Behavior

Configuration Analysis

Original Timer Configuration

Service Configuration

Root Cause Analysis

Systemd Dependency Resolution

Systemd Timer Mechanics

Solution

Implementation

Verification

Related Issues

Common Misconceptions

Alternative Diagnostic Approaches

Timer State Management

Technical References

Conclusion

Displaying Google Calendar Events in Conky

Why Use the API Instead of iCal?

Prerequisites

Setup Process

1. Create Service Account

2. Grant Calendar Access

3. Project Structure

4. Configuration

5. Python Script

6. Conky Integration

Testing

Key Features

Configuration Options

Troubleshooting

Security Considerations

Limitations

Image location detection using Bedrock and Claude Model

Overview

Setup

Image Processing

Base64 Encoding

Bedrock Integration

File Management

Main Processing Loop

Usage

Output Example

Technical Notes

Automating Gmail PDF Extraction and Paperless NGX Integration

Features

Architecture

Installation

Prerequisites

Configuration

Gmail App Password

Core Script

PDF Processor Module

Paperless NGX Uploader

Running Paperless NGX in Docker

Docker Compose Configuration

Environment Configuration

Installation Steps

Accessing Paperless

Mobile Access with Paperless NGX Android App

Automation with Systemd

Service File (~/.config/systemd/user/gmail-pdf-processor.service)

Timer File (~/.config/systemd/user/gmail-pdf-processor.timer)

Enable and Start

Benefits

Security Considerations

Conclusion

Building a RAG System for Your Jekyll Blog with AWS Bedrock Claude

What We’re Building

Architecture Overview

Why This Approach Works

Prerequisites

Service File (`~/.config/systemd/user/gmail-pdf-processor.service`)

Timer File (`~/.config/systemd/user/gmail-pdf-processor.timer`)